1
0

FramingIface.c 5.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172
  1. /* vim: set expandtab ts=4 sw=4: */
  2. /*
  3. * You may redistribute this program and/or modify it under the terms of
  4. * the GNU General Public License as published by the Free Software Foundation,
  5. * either version 3 of the License, or (at your option) any later version.
  6. *
  7. * This program is distributed in the hope that it will be useful,
  8. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  9. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  10. * GNU General Public License for more details.
  11. *
  12. * You should have received a copy of the GNU General Public License
  13. * along with this program. If not, see <http://www.gnu.org/licenses/>.
  14. */
  15. #include "interface/Iface.h"
  16. #include "memory/Allocator.h"
  17. #include "util/Identity.h"
  18. #include "wire/Error.h"
  19. #define REQUIRED_PADDING 512
  20. struct MessageList;
  21. struct MessageList {
  22. struct Message* msg;
  23. struct MessageList* next;
  24. };
  25. struct FramingIface_pvt {
  26. struct Iface messageIf;
  27. struct Iface streamIf;
  28. const uint32_t maxMessageSize;
  29. struct Allocator* alloc;
  30. // fields specific to this frame.
  31. uint32_t bytesRemaining;
  32. struct Allocator* frameAlloc;
  33. struct MessageList* frameParts;
  34. union {
  35. uint32_t length_be;
  36. uint8_t bytes[4];
  37. } header;
  38. uint32_t headerIndex;
  39. Identity
  40. };
  41. static struct Message* mergeMessage(struct FramingIface_pvt* fi, struct Message* last)
  42. {
  43. int length = last->length;
  44. // The only accurate way to get the full length because this last might contain
  45. // the beginning of the next frame.
  46. struct MessageList* part = fi->frameParts;
  47. for (part = fi->frameParts; part; part = part->next) {
  48. length += part->msg->length;
  49. }
  50. struct Message* out = Message_new(0, length + REQUIRED_PADDING, fi->frameAlloc);
  51. Message_push(out, last->bytes, last->length, NULL);
  52. for (part = fi->frameParts; part; part = part->next) {
  53. Message_push(out, part->msg->bytes, part->msg->length, NULL);
  54. }
  55. Assert_true(length <= out->length);
  56. return out;
  57. }
  58. static Iface_DEFUN receiveMessage(struct Message* msg, struct Iface* streamIf)
  59. {
  60. struct FramingIface_pvt* fi = Identity_containerOf(streamIf, struct FramingIface_pvt, streamIf);
  61. if (fi->bytesRemaining > fi->maxMessageSize) {
  62. // Oversize message
  63. Assert_true(0);
  64. return NULL;
  65. }
  66. if (fi->frameParts) {
  67. if (fi->bytesRemaining <= (uint32_t)msg->length) {
  68. struct Message* wholeMessage = mergeMessage(fi, msg);
  69. fi->bytesRemaining = 0;
  70. fi->frameParts = NULL;
  71. Assert_true(fi->headerIndex == 0);
  72. struct Allocator* frameAlloc = fi->frameAlloc;
  73. fi->frameAlloc = NULL;
  74. // Run the message through again since it's almost certainly not perfect size.
  75. Iface_CALL(receiveMessage, wholeMessage, streamIf);
  76. Allocator_free(frameAlloc);
  77. return NULL;
  78. }
  79. fi->bytesRemaining -= msg->length;
  80. Allocator_adopt(fi->frameAlloc, msg->alloc);
  81. struct MessageList* parts = Allocator_malloc(fi->frameAlloc, sizeof(struct MessageList));
  82. parts->msg = msg;
  83. parts->next = fi->frameParts;
  84. fi->frameParts = parts;
  85. return NULL;
  86. }
  87. for (;;) {
  88. while (fi->headerIndex < 4) {
  89. if (!msg->length) {
  90. return NULL;
  91. }
  92. fi->header.bytes[fi->headerIndex] = msg->bytes[0];
  93. Message_shift(msg, -1, NULL);
  94. fi->headerIndex++;
  95. }
  96. fi->headerIndex = 0;
  97. fi->bytesRemaining = Endian_bigEndianToHost32(fi->header.length_be);
  98. if (fi->bytesRemaining > fi->maxMessageSize) {
  99. // oversize
  100. Assert_true(0);
  101. return NULL;
  102. }
  103. if (fi->bytesRemaining == (uint32_t)msg->length) {
  104. fi->bytesRemaining = 0;
  105. return Iface_next(&fi->messageIf, msg);
  106. } else if (fi->bytesRemaining < (uint32_t)msg->length) {
  107. struct Allocator* alloc = Allocator_child(msg->alloc);
  108. struct Message* m = Message_new(fi->bytesRemaining, REQUIRED_PADDING, alloc);
  109. Bits_memcpy(m->bytes, msg->bytes, fi->bytesRemaining);
  110. Message_shift(msg, -fi->bytesRemaining, NULL);
  111. fi->bytesRemaining = 0;
  112. Iface_send(&fi->messageIf, m);
  113. Allocator_free(alloc);
  114. continue;
  115. } else {
  116. fi->frameAlloc = Allocator_child(fi->alloc);
  117. struct Message* m = Allocator_calloc(fi->frameAlloc, sizeof(struct Message), 1);
  118. m->capacity = m->length = msg->length + 4;
  119. m->bytes = Allocator_malloc(fi->frameAlloc, m->length);
  120. m->alloc = fi->frameAlloc;
  121. Message_shift(m, -m->length, NULL);
  122. Message_push(m, msg->bytes, msg->length, NULL);
  123. Message_push(m, fi->header.bytes, 4, NULL);
  124. fi->bytesRemaining -= msg->length;
  125. fi->frameParts = Allocator_malloc(fi->frameAlloc, sizeof(struct MessageList));
  126. fi->frameParts->msg = m;
  127. fi->frameParts->next = NULL;
  128. }
  129. return NULL;
  130. }
  131. }
  132. static Iface_DEFUN sendMessage(struct Message* msg, struct Iface* messageIf)
  133. {
  134. struct FramingIface_pvt* fi =
  135. Identity_containerOf(messageIf, struct FramingIface_pvt, messageIf);
  136. Message_push32(msg, msg->length, NULL);
  137. return Iface_next(&fi->streamIf, msg);
  138. }
  139. struct Iface* FramingIface_new(uint32_t maxMsgSize, struct Iface* toWrap, struct Allocator* alloc)
  140. {
  141. struct FramingIface_pvt* context =
  142. Allocator_clone(alloc, (&(struct FramingIface_pvt) {
  143. .maxMessageSize = maxMsgSize,
  144. .alloc = alloc,
  145. .streamIf = { .send = receiveMessage },
  146. .messageIf = { .send = sendMessage }
  147. }));
  148. Identity_set(context);
  149. Iface_plumb(toWrap, &context->streamIf);
  150. return &context->messageIf;
  151. }