Hermes.c 6.9 KB

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