1
0

Hermes.c 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211
  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 "admin/angel/Hermes.h"
  16. #include "benc/Dict.h"
  17. #include "benc/serialization/BencSerializer.h"
  18. #include "benc/serialization/standard/StandardBencSerializer.h"
  19. #include "memory/Allocator.h"
  20. #include "interface/Interface.h"
  21. #include "io/Reader.h"
  22. #include "io/ArrayReader.h"
  23. #include "io/Writer.h"
  24. #include "io/ArrayWriter.h"
  25. #include "util/events/Event.h"
  26. #include "util/events/EventBase.h"
  27. #include "util/log/Log.h"
  28. #include "util/events/Timeout.h"
  29. #include "util/Identity.h"
  30. #include "util/Hex.h"
  31. #define REQ_TIMEOUT 10000
  32. struct Request
  33. {
  34. struct Allocator* alloc;
  35. Hermes_onResponse onResponse;
  36. void* onResponseContext;
  37. struct Hermes* hermes;
  38. uint32_t handle;
  39. Identity
  40. };
  41. #define Map_NAME RequestSet
  42. #define Map_VALUE_TYPE struct Request*
  43. #define Map_ENABLE_HANDLES
  44. #include "util/Map.h"
  45. struct Hermes
  46. {
  47. struct Interface* iface;
  48. struct Allocator* alloc;
  49. struct EventBase* eventBase;
  50. struct Map_RequestSet requestSet;
  51. struct Log* logger;
  52. Identity
  53. };
  54. /** Called when the request allocator is freed. */
  55. static int removeReqFromSet(struct Allocator_OnFreeJob* job)
  56. {
  57. struct Request* req = Identity_cast((struct Request*) job->userData);
  58. struct Hermes* h = Identity_cast(req->hermes);
  59. int index = Map_RequestSet_indexForHandle(req->handle, &h->requestSet);
  60. if (index > -1) {
  61. Map_RequestSet_remove(index, &h->requestSet);
  62. } else {
  63. Log_error(h->logger, "request with handle [%u] missing from set", req->handle);
  64. }
  65. return 0;
  66. }
  67. static void timeout(void* vrequest)
  68. {
  69. struct Request* req = Identity_cast((struct Request*) vrequest);
  70. Dict resp = Dict_CONST(String_CONST("error"), String_OBJ(String_CONST("timeout")), NULL);
  71. req->onResponse(&resp, req->onResponseContext);
  72. Allocator_free(req->alloc);
  73. }
  74. static void receiveMessage2(struct Message* msg, struct Hermes* hermes, struct Allocator* tempAlloc)
  75. {
  76. #ifdef Log_KEYS
  77. char lastChr = msg->bytes[msg->length - 1];
  78. msg->bytes[msg->length - 1] = '\0';
  79. Log_keys(hermes->logger, "Got message from angel [%s%c]", msg->bytes, lastChr);
  80. msg->bytes[msg->length - 1] = lastChr;
  81. #else
  82. Log_debug(hermes->logger, "Got message from angel");
  83. #endif
  84. struct Reader* reader = ArrayReader_new(msg->bytes, msg->length, tempAlloc);
  85. Dict d;
  86. if (StandardBencSerializer_get()->parseDictionary(reader, tempAlloc, &d)) {
  87. Log_warn(hermes->logger, "Failed to parse message from angel");
  88. return;
  89. }
  90. String* txid = Dict_getString(&d, String_CONST("txid"));
  91. uint32_t handle;
  92. if (!txid || txid->len != 8 || 4 != Hex_decode((uint8_t*)&handle, 4, (uint8_t*)txid->bytes, 8))
  93. {
  94. Log_warn(hermes->logger, "Message from angel; txid missing or unrecognized");
  95. return;
  96. }
  97. int index = Map_RequestSet_indexForHandle(handle, &hermes->requestSet);
  98. if (index < 0) {
  99. Log_warn(hermes->logger, "Message from angel references nonexistant request");
  100. return;
  101. }
  102. struct Request* req = Identity_cast((struct Request*) hermes->requestSet.values[index]);
  103. req->onResponse(&d, req->onResponseContext);
  104. Allocator_free(req->alloc);
  105. }
  106. static uint8_t receiveMessage(struct Message* msg, struct Interface* iface)
  107. {
  108. struct Hermes* hermes = Identity_cast((struct Hermes*) iface->receiverContext);
  109. struct Allocator* alloc = Allocator_child(hermes->alloc);
  110. receiveMessage2(msg, hermes, alloc);
  111. Allocator_free(alloc);
  112. return 0;
  113. }
  114. void Hermes_callAngel(Dict* message,
  115. Hermes_onResponse onResponse,
  116. void* onResponseContext,
  117. struct Allocator* alloc,
  118. struct Except* eh,
  119. struct Hermes* hermes)
  120. {
  121. Identity_check(hermes);
  122. #define PADDING 32
  123. #define BUFF_SIZE 1024
  124. struct {
  125. uint8_t padding[PADDING];
  126. uint8_t message[BUFF_SIZE];
  127. } buff = { .padding = {0}, .message = {0} };
  128. struct Allocator* reqAlloc = Allocator_child(alloc);
  129. struct Request* req = Allocator_clone(reqAlloc, (&(struct Request) {
  130. .alloc = reqAlloc,
  131. .onResponse = onResponse,
  132. .onResponseContext = onResponseContext,
  133. .hermes = hermes
  134. }));
  135. Identity_set(req);
  136. int index = Map_RequestSet_put(&req, &hermes->requestSet);
  137. Allocator_onFree(reqAlloc, removeReqFromSet, req);
  138. uint32_t handle = hermes->requestSet.handles[index];
  139. req->handle = handle;
  140. uint8_t handleHex[9];
  141. Hex_encode(handleHex, 9, (uint8_t*)&handle, 4);
  142. Dict_putString(message, String_CONST("txid"), String_CONST((char*)handleHex), reqAlloc);
  143. struct Writer* writer = ArrayWriter_new(buff.message, BUFF_SIZE, reqAlloc);
  144. if (StandardBencSerializer_get()->serializeDictionary(writer, message)) {
  145. Except_raise(eh, Hermes_callAngel_ESERIALIZE, "Failed to serialize message");
  146. }
  147. // Remove the txid string so there is not a dangling pointer in the message.
  148. Dict_remove(message, String_CONST("txid"));
  149. // This is done in a strange way but it is to prevent "possible buffer overflow" errors
  150. struct Message* m = &(struct Message) {
  151. .bytes = (uint8_t*) &buff,
  152. .length = writer->bytesWritten + PADDING,
  153. .padding = 0
  154. };
  155. m->capacity = m->length;
  156. Message_shift(m, -PADDING);
  157. Log_debug(hermes->logger, "Sending [%d] bytes to angel [%s].", m->length, m->bytes);
  158. m = Message_clone(m, reqAlloc);
  159. int ret = Interface_sendMessage(hermes->iface, m);
  160. if (ret) {
  161. Except_raise(eh, Hermes_callAngel_ESEND, "Failed to send message to angel [%d]", ret);
  162. }
  163. // Use interval as defensive programming
  164. // the Allocator_free() in the timeout callback deactivates it.
  165. Timeout_setInterval(timeout, req, REQ_TIMEOUT, hermes->eventBase, reqAlloc);
  166. }
  167. struct Hermes* Hermes_new(struct Interface* angelIface,
  168. struct EventBase* eventBase,
  169. struct Log* logger,
  170. struct Allocator* alloc)
  171. {
  172. struct Hermes* out = Allocator_clone(alloc, (&(struct Hermes) {
  173. .iface = angelIface,
  174. .alloc = alloc,
  175. .eventBase = eventBase,
  176. .requestSet = {
  177. .allocator = alloc
  178. },
  179. .logger = logger
  180. }));
  181. Identity_set(out);
  182. angelIface->receiveMessage = receiveMessage;
  183. angelIface->receiverContext = out;
  184. return out;
  185. }