observers.c 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264
  1. /*
  2. *
  3. * Copyright 2016, Google Inc.
  4. * All rights reserved.
  5. *
  6. * Redistribution and use in source and binary forms, with or without
  7. * modification, are permitted provided that the following conditions are
  8. * met:
  9. *
  10. * * Redistributions of source code must retain the above copyright
  11. * notice, this list of conditions and the following disclaimer.
  12. * * Redistributions in binary form must reproduce the above
  13. * copyright notice, this list of conditions and the following disclaimer
  14. * in the documentation and/or other materials provided with the
  15. * distribution.
  16. * * Neither the name of Google Inc. nor the names of its
  17. * contributors may be used to endorse or promote products derived from
  18. * this software without specific prior written permission.
  19. *
  20. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  21. * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  22. * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  23. * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  24. * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  25. * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  26. * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  27. * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  28. * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  29. * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  30. * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  31. *
  32. */
  33. #include "internal.h"
  34. #include "cgrpc.h"
  35. #include <stdlib.h>
  36. #include <string.h>
  37. // create observers for each type of GRPC operation
  38. cgrpc_observer_send_initial_metadata *cgrpc_observer_create_send_initial_metadata(cgrpc_metadata_array *metadata) {
  39. cgrpc_observer_send_initial_metadata *observer =
  40. (cgrpc_observer_send_initial_metadata *) malloc(sizeof (cgrpc_observer_send_initial_metadata));
  41. observer->op_type = GRPC_OP_SEND_INITIAL_METADATA;
  42. cgrpc_metadata_array_move_metadata(&(observer->initial_metadata), metadata);
  43. return observer;
  44. }
  45. cgrpc_observer_send_message *cgrpc_observer_create_send_message() {
  46. cgrpc_observer_send_message *observer =
  47. (cgrpc_observer_send_message *) malloc(sizeof (cgrpc_observer_send_message));
  48. observer->op_type = GRPC_OP_SEND_MESSAGE;
  49. return observer;
  50. }
  51. cgrpc_observer_send_close_from_client *cgrpc_observer_create_send_close_from_client() {
  52. cgrpc_observer_send_close_from_client *observer =
  53. (cgrpc_observer_send_close_from_client *) malloc(sizeof (cgrpc_observer_send_close_from_client));
  54. observer->op_type = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
  55. return observer;
  56. }
  57. cgrpc_observer_send_status_from_server *cgrpc_observer_create_send_status_from_server(cgrpc_metadata_array *metadata) {
  58. cgrpc_observer_send_status_from_server *observer =
  59. (cgrpc_observer_send_status_from_server *) malloc(sizeof (cgrpc_observer_send_status_from_server));
  60. observer->op_type = GRPC_OP_SEND_STATUS_FROM_SERVER;
  61. cgrpc_metadata_array_move_metadata(&(observer->trailing_metadata), metadata);
  62. return observer;
  63. }
  64. cgrpc_observer_recv_initial_metadata *cgrpc_observer_create_recv_initial_metadata() {
  65. cgrpc_observer_recv_initial_metadata *observer =
  66. (cgrpc_observer_recv_initial_metadata *) malloc(sizeof (cgrpc_observer_recv_initial_metadata));
  67. observer->op_type = GRPC_OP_RECV_INITIAL_METADATA;
  68. return observer;
  69. }
  70. cgrpc_observer_recv_message *cgrpc_observer_create_recv_message() {
  71. cgrpc_observer_recv_message *observer =
  72. (cgrpc_observer_recv_message *) malloc(sizeof (cgrpc_observer_recv_message));
  73. observer->op_type = GRPC_OP_RECV_MESSAGE;
  74. observer->response_payload_recv = NULL;
  75. return observer;
  76. }
  77. cgrpc_observer_recv_status_on_client *cgrpc_observer_create_recv_status_on_client() {
  78. cgrpc_observer_recv_status_on_client *observer =
  79. (cgrpc_observer_recv_status_on_client *) malloc(sizeof (cgrpc_observer_recv_status_on_client));
  80. observer->op_type = GRPC_OP_RECV_STATUS_ON_CLIENT;
  81. return observer;
  82. }
  83. cgrpc_observer_recv_close_on_server *cgrpc_observer_create_recv_close_on_server() {
  84. cgrpc_observer_recv_close_on_server *observer =
  85. (cgrpc_observer_recv_close_on_server *) malloc(sizeof (cgrpc_observer_recv_close_on_server));
  86. observer->op_type = GRPC_OP_RECV_CLOSE_ON_SERVER;
  87. observer->was_cancelled = 0;
  88. return observer;
  89. }
  90. // apply observer to operation
  91. void cgrpc_observer_apply(cgrpc_observer *observer, grpc_op *op) {
  92. op->op = observer->op_type;
  93. op->flags = 0;
  94. op->reserved = NULL;
  95. switch (observer->op_type) {
  96. case GRPC_OP_SEND_INITIAL_METADATA: {
  97. cgrpc_observer_send_initial_metadata *obs = (cgrpc_observer_send_initial_metadata *) observer;
  98. op->data.send_initial_metadata.count = obs->initial_metadata.count;
  99. op->data.send_initial_metadata.metadata = obs->initial_metadata.metadata;
  100. break;
  101. }
  102. case GRPC_OP_SEND_MESSAGE: {
  103. cgrpc_observer_send_message *obs = (cgrpc_observer_send_message *) observer;
  104. op->data.send_message = obs->request_payload;
  105. break;
  106. }
  107. case GRPC_OP_SEND_CLOSE_FROM_CLIENT: {
  108. break;
  109. }
  110. case GRPC_OP_SEND_STATUS_FROM_SERVER: {
  111. cgrpc_observer_send_status_from_server *obs = (cgrpc_observer_send_status_from_server *) observer;
  112. op->data.send_status_from_server.trailing_metadata_count = obs->trailing_metadata.count;
  113. op->data.send_status_from_server.trailing_metadata = obs->trailing_metadata.metadata;
  114. op->data.send_status_from_server.status = obs->status;
  115. op->data.send_status_from_server.status_details = obs->status_details;
  116. break;
  117. }
  118. case GRPC_OP_RECV_INITIAL_METADATA: {
  119. cgrpc_observer_recv_initial_metadata *obs = (cgrpc_observer_recv_initial_metadata *) observer;
  120. grpc_metadata_array_init(&(obs->initial_metadata_recv));
  121. op->data.recv_initial_metadata = &(obs->initial_metadata_recv);
  122. break;
  123. }
  124. case GRPC_OP_RECV_MESSAGE: {
  125. cgrpc_observer_recv_message *obs = (cgrpc_observer_recv_message *) observer;
  126. op->data.recv_message = &(obs->response_payload_recv);
  127. break;
  128. }
  129. case GRPC_OP_RECV_STATUS_ON_CLIENT: {
  130. cgrpc_observer_recv_status_on_client *obs = (cgrpc_observer_recv_status_on_client *) observer;
  131. grpc_metadata_array_init(&(obs->trailing_metadata_recv));
  132. obs->server_status = GRPC_STATUS_OK;
  133. obs->server_details = NULL;
  134. obs->server_details_capacity = 0;
  135. op->data.recv_status_on_client.trailing_metadata = &(obs->trailing_metadata_recv);
  136. op->data.recv_status_on_client.status = &(obs->server_status);
  137. op->data.recv_status_on_client.status_details = &(obs->server_details);
  138. op->data.recv_status_on_client.status_details_capacity = &(obs->server_details_capacity);
  139. break;
  140. }
  141. case GRPC_OP_RECV_CLOSE_ON_SERVER: {
  142. cgrpc_observer_recv_close_on_server *obs = (cgrpc_observer_recv_close_on_server *) observer;
  143. op->data.recv_close_on_server.cancelled = &(obs->was_cancelled);
  144. break;
  145. }
  146. default: {
  147. }
  148. }
  149. }
  150. // destroy all observers
  151. void cgrpc_observer_destroy(cgrpc_observer *observer) {
  152. switch (observer->op_type) {
  153. case GRPC_OP_SEND_INITIAL_METADATA: {
  154. cgrpc_observer_send_initial_metadata *obs = (cgrpc_observer_send_initial_metadata *) observer;
  155. grpc_metadata_array_destroy(&(obs->initial_metadata));
  156. free(obs);
  157. break;
  158. }
  159. case GRPC_OP_SEND_MESSAGE: {
  160. cgrpc_observer_send_message *obs = (cgrpc_observer_send_message *) observer;
  161. grpc_byte_buffer_destroy(obs->request_payload);
  162. free(obs);
  163. break;
  164. }
  165. case GRPC_OP_SEND_CLOSE_FROM_CLIENT: {
  166. cgrpc_observer_send_close_from_client *obs = (cgrpc_observer_send_close_from_client *) observer;
  167. free(obs);
  168. break;
  169. }
  170. case GRPC_OP_SEND_STATUS_FROM_SERVER: {
  171. cgrpc_observer_send_status_from_server *obs = (cgrpc_observer_send_status_from_server *) observer;
  172. free(obs);
  173. break;
  174. }
  175. case GRPC_OP_RECV_INITIAL_METADATA: {
  176. cgrpc_observer_recv_initial_metadata *obs = (cgrpc_observer_recv_initial_metadata *) observer;
  177. grpc_metadata_array_destroy(&obs->initial_metadata_recv);
  178. free(obs);
  179. break;
  180. }
  181. case GRPC_OP_RECV_MESSAGE: {
  182. cgrpc_observer_recv_message *obs = (cgrpc_observer_recv_message *) observer;
  183. grpc_byte_buffer_destroy(obs->response_payload_recv);
  184. free(obs);
  185. break;
  186. }
  187. case GRPC_OP_RECV_STATUS_ON_CLIENT: {
  188. cgrpc_observer_recv_status_on_client *obs = (cgrpc_observer_recv_status_on_client *) observer;
  189. free(obs->server_details);
  190. grpc_metadata_array_destroy(&(obs->trailing_metadata_recv));
  191. free(obs);
  192. break;
  193. }
  194. case GRPC_OP_RECV_CLOSE_ON_SERVER: {
  195. cgrpc_observer_recv_close_on_server *obs = (cgrpc_observer_recv_close_on_server *) observer;
  196. free(obs);
  197. break;
  198. }
  199. default: {
  200. }
  201. }
  202. }
  203. cgrpc_byte_buffer *cgrpc_observer_recv_message_get_message(cgrpc_observer_recv_message *observer) {
  204. if (observer->response_payload_recv) {
  205. return grpc_byte_buffer_copy(observer->response_payload_recv);
  206. } else {
  207. return NULL;
  208. }
  209. }
  210. cgrpc_metadata_array *cgrpc_observer_recv_initial_metadata_get_metadata(cgrpc_observer_recv_initial_metadata *observer) {
  211. cgrpc_metadata_array *metadata = cgrpc_metadata_array_create();
  212. cgrpc_metadata_array_move_metadata(metadata, &(observer->initial_metadata_recv));
  213. return metadata;
  214. }
  215. void cgrpc_observer_send_message_set_message(cgrpc_observer_send_message *observer, cgrpc_byte_buffer *message) {
  216. observer->request_payload = grpc_byte_buffer_copy(message);
  217. }
  218. void cgrpc_observer_send_status_from_server_set_status(cgrpc_observer_send_status_from_server *observer, int status) {
  219. observer->status = status;
  220. }
  221. void cgrpc_observer_send_status_from_server_set_status_details(cgrpc_observer_send_status_from_server *observer, const char *statusDetails) {
  222. observer->status_details = strdup(statusDetails);
  223. }
  224. cgrpc_metadata_array *cgrpc_observer_recv_status_on_client_get_metadata(cgrpc_observer_recv_status_on_client *observer) {
  225. cgrpc_metadata_array *metadata = cgrpc_metadata_array_create();
  226. cgrpc_metadata_array_move_metadata(metadata, &(observer->trailing_metadata_recv));
  227. return metadata;
  228. }
  229. long cgrpc_observer_recv_status_on_client_get_status(cgrpc_observer_recv_status_on_client *observer) {
  230. return observer->server_status;
  231. }
  232. const char *cgrpc_observer_recv_status_on_client_get_status_details(cgrpc_observer_recv_status_on_client *observer) {
  233. return observer->server_details;
  234. }
  235. int cgrpc_observer_recv_close_on_server_get_was_cancelled(cgrpc_observer_recv_close_on_server *observer) {
  236. return observer->was_cancelled;
  237. }