observers_shim.c 9.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250
  1. /*
  2. * Copyright 2016, gRPC Authors All rights reserved.
  3. *
  4. * Licensed under the Apache License, Version 2.0 (the "License");
  5. * you may not use this file except in compliance with the License.
  6. * You may obtain a copy of the License at
  7. *
  8. * http://www.apache.org/licenses/LICENSE-2.0
  9. *
  10. * Unless required by applicable law or agreed to in writing, software
  11. * distributed under the License is distributed on an "AS IS" BASIS,
  12. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. * See the License for the specific language governing permissions and
  14. * limitations under the License.
  15. */
  16. #include "internal.h"
  17. #include "cgrpc.h"
  18. #include <stdlib.h>
  19. #include <string.h>
  20. // create observers for each type of GRPC operation
  21. cgrpc_observer_send_initial_metadata *cgrpc_observer_create_send_initial_metadata(cgrpc_metadata_array *metadata) {
  22. cgrpc_observer_send_initial_metadata *observer =
  23. (cgrpc_observer_send_initial_metadata *) malloc(sizeof (cgrpc_observer_send_initial_metadata));
  24. observer->op_type = GRPC_OP_SEND_INITIAL_METADATA;
  25. cgrpc_metadata_array_move_metadata(&(observer->initial_metadata), metadata);
  26. return observer;
  27. }
  28. cgrpc_observer_send_message *cgrpc_observer_create_send_message() {
  29. cgrpc_observer_send_message *observer =
  30. (cgrpc_observer_send_message *) malloc(sizeof (cgrpc_observer_send_message));
  31. observer->op_type = GRPC_OP_SEND_MESSAGE;
  32. return observer;
  33. }
  34. cgrpc_observer_send_close_from_client *cgrpc_observer_create_send_close_from_client() {
  35. cgrpc_observer_send_close_from_client *observer =
  36. (cgrpc_observer_send_close_from_client *) malloc(sizeof (cgrpc_observer_send_close_from_client));
  37. observer->op_type = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
  38. return observer;
  39. }
  40. cgrpc_observer_send_status_from_server *cgrpc_observer_create_send_status_from_server(cgrpc_metadata_array *metadata) {
  41. cgrpc_observer_send_status_from_server *observer =
  42. (cgrpc_observer_send_status_from_server *) malloc(sizeof (cgrpc_observer_send_status_from_server));
  43. observer->op_type = GRPC_OP_SEND_STATUS_FROM_SERVER;
  44. cgrpc_metadata_array_move_metadata(&(observer->trailing_metadata), metadata);
  45. return observer;
  46. }
  47. cgrpc_observer_recv_initial_metadata *cgrpc_observer_create_recv_initial_metadata() {
  48. cgrpc_observer_recv_initial_metadata *observer =
  49. (cgrpc_observer_recv_initial_metadata *) malloc(sizeof (cgrpc_observer_recv_initial_metadata));
  50. observer->op_type = GRPC_OP_RECV_INITIAL_METADATA;
  51. return observer;
  52. }
  53. cgrpc_observer_recv_message *cgrpc_observer_create_recv_message() {
  54. cgrpc_observer_recv_message *observer =
  55. (cgrpc_observer_recv_message *) malloc(sizeof (cgrpc_observer_recv_message));
  56. observer->op_type = GRPC_OP_RECV_MESSAGE;
  57. observer->response_payload_recv = NULL;
  58. return observer;
  59. }
  60. cgrpc_observer_recv_status_on_client *cgrpc_observer_create_recv_status_on_client() {
  61. cgrpc_observer_recv_status_on_client *observer =
  62. (cgrpc_observer_recv_status_on_client *) malloc(sizeof (cgrpc_observer_recv_status_on_client));
  63. observer->op_type = GRPC_OP_RECV_STATUS_ON_CLIENT;
  64. return observer;
  65. }
  66. cgrpc_observer_recv_close_on_server *cgrpc_observer_create_recv_close_on_server() {
  67. cgrpc_observer_recv_close_on_server *observer =
  68. (cgrpc_observer_recv_close_on_server *) malloc(sizeof (cgrpc_observer_recv_close_on_server));
  69. observer->op_type = GRPC_OP_RECV_CLOSE_ON_SERVER;
  70. observer->was_cancelled = 0;
  71. return observer;
  72. }
  73. // apply observer to operation
  74. void cgrpc_observer_apply(cgrpc_observer *observer, grpc_op *op) {
  75. op->op = observer->op_type;
  76. op->flags = 0;
  77. op->reserved = NULL;
  78. switch (observer->op_type) {
  79. case GRPC_OP_SEND_INITIAL_METADATA: {
  80. cgrpc_observer_send_initial_metadata *obs = (cgrpc_observer_send_initial_metadata *) observer;
  81. op->data.send_initial_metadata.count = obs->initial_metadata.count;
  82. op->data.send_initial_metadata.metadata = obs->initial_metadata.metadata;
  83. break;
  84. }
  85. case GRPC_OP_SEND_MESSAGE: {
  86. cgrpc_observer_send_message *obs = (cgrpc_observer_send_message *) observer;
  87. op->data.send_message.send_message = obs->request_payload;
  88. break;
  89. }
  90. case GRPC_OP_SEND_CLOSE_FROM_CLIENT: {
  91. break;
  92. }
  93. case GRPC_OP_SEND_STATUS_FROM_SERVER: {
  94. cgrpc_observer_send_status_from_server *obs = (cgrpc_observer_send_status_from_server *) observer;
  95. op->data.send_status_from_server.trailing_metadata_count = obs->trailing_metadata.count;
  96. op->data.send_status_from_server.trailing_metadata = obs->trailing_metadata.metadata;
  97. op->data.send_status_from_server.status = obs->status;
  98. op->data.send_status_from_server.status_details = &obs->status_details;
  99. break;
  100. }
  101. case GRPC_OP_RECV_INITIAL_METADATA: {
  102. cgrpc_observer_recv_initial_metadata *obs = (cgrpc_observer_recv_initial_metadata *) observer;
  103. grpc_metadata_array_init(&(obs->initial_metadata_recv));
  104. op->data.recv_initial_metadata.recv_initial_metadata = &(obs->initial_metadata_recv);
  105. break;
  106. }
  107. case GRPC_OP_RECV_MESSAGE: {
  108. cgrpc_observer_recv_message *obs = (cgrpc_observer_recv_message *) observer;
  109. op->data.recv_message.recv_message = &(obs->response_payload_recv);
  110. break;
  111. }
  112. case GRPC_OP_RECV_STATUS_ON_CLIENT: {
  113. cgrpc_observer_recv_status_on_client *obs = (cgrpc_observer_recv_status_on_client *) observer;
  114. grpc_metadata_array_init(&(obs->trailing_metadata_recv));
  115. obs->server_status = GRPC_STATUS_OK;
  116. obs->server_details = grpc_slice_from_copied_string("");
  117. op->data.recv_status_on_client.trailing_metadata = &(obs->trailing_metadata_recv);
  118. op->data.recv_status_on_client.status = &(obs->server_status);
  119. op->data.recv_status_on_client.status_details = &(obs->server_details);
  120. break;
  121. }
  122. case GRPC_OP_RECV_CLOSE_ON_SERVER: {
  123. cgrpc_observer_recv_close_on_server *obs = (cgrpc_observer_recv_close_on_server *) observer;
  124. op->data.recv_close_on_server.cancelled = &(obs->was_cancelled);
  125. break;
  126. }
  127. default: {
  128. }
  129. }
  130. }
  131. // destroy all observers
  132. void cgrpc_observer_destroy(cgrpc_observer *observer) {
  133. switch (observer->op_type) {
  134. case GRPC_OP_SEND_INITIAL_METADATA: {
  135. cgrpc_observer_send_initial_metadata *obs = (cgrpc_observer_send_initial_metadata *) observer;
  136. grpc_metadata_array_destroy(&(obs->initial_metadata));
  137. free(obs);
  138. break;
  139. }
  140. case GRPC_OP_SEND_MESSAGE: {
  141. cgrpc_observer_send_message *obs = (cgrpc_observer_send_message *) observer;
  142. grpc_byte_buffer_destroy(obs->request_payload);
  143. free(obs);
  144. break;
  145. }
  146. case GRPC_OP_SEND_CLOSE_FROM_CLIENT: {
  147. cgrpc_observer_send_close_from_client *obs = (cgrpc_observer_send_close_from_client *) observer;
  148. free(obs);
  149. break;
  150. }
  151. case GRPC_OP_SEND_STATUS_FROM_SERVER: {
  152. cgrpc_observer_send_status_from_server *obs = (cgrpc_observer_send_status_from_server *) observer;
  153. grpc_slice_unref(obs->status_details);
  154. free(obs);
  155. break;
  156. }
  157. case GRPC_OP_RECV_INITIAL_METADATA: {
  158. cgrpc_observer_recv_initial_metadata *obs = (cgrpc_observer_recv_initial_metadata *) observer;
  159. grpc_metadata_array_destroy(&obs->initial_metadata_recv);
  160. free(obs);
  161. break;
  162. }
  163. case GRPC_OP_RECV_MESSAGE: {
  164. cgrpc_observer_recv_message *obs = (cgrpc_observer_recv_message *) observer;
  165. grpc_byte_buffer_destroy(obs->response_payload_recv);
  166. free(obs);
  167. break;
  168. }
  169. case GRPC_OP_RECV_STATUS_ON_CLIENT: {
  170. cgrpc_observer_recv_status_on_client *obs = (cgrpc_observer_recv_status_on_client *) observer;
  171. grpc_metadata_array_destroy(&(obs->trailing_metadata_recv));
  172. grpc_slice_unref(obs->server_details);
  173. free(obs);
  174. break;
  175. }
  176. case GRPC_OP_RECV_CLOSE_ON_SERVER: {
  177. cgrpc_observer_recv_close_on_server *obs = (cgrpc_observer_recv_close_on_server *) observer;
  178. free(obs);
  179. break;
  180. }
  181. default: {
  182. }
  183. }
  184. }
  185. cgrpc_byte_buffer *cgrpc_observer_recv_message_get_message(cgrpc_observer_recv_message *observer) {
  186. if (observer->response_payload_recv) {
  187. return grpc_byte_buffer_copy(observer->response_payload_recv);
  188. } else {
  189. return NULL;
  190. }
  191. }
  192. cgrpc_metadata_array *cgrpc_observer_recv_initial_metadata_get_metadata(cgrpc_observer_recv_initial_metadata *observer) {
  193. cgrpc_metadata_array *metadata = cgrpc_metadata_array_create();
  194. cgrpc_metadata_array_move_metadata(metadata, &(observer->initial_metadata_recv));
  195. return metadata;
  196. }
  197. void cgrpc_observer_send_message_set_message(cgrpc_observer_send_message *observer, cgrpc_byte_buffer *message) {
  198. observer->request_payload = grpc_byte_buffer_copy(message);
  199. }
  200. void cgrpc_observer_send_status_from_server_set_status(cgrpc_observer_send_status_from_server *observer, int status) {
  201. observer->status = status;
  202. }
  203. void cgrpc_observer_send_status_from_server_set_status_details(cgrpc_observer_send_status_from_server *observer, const char *statusDetails) {
  204. observer->status_details = grpc_slice_from_copied_string(statusDetails);
  205. }
  206. cgrpc_metadata_array *cgrpc_observer_recv_status_on_client_get_metadata(cgrpc_observer_recv_status_on_client *observer) {
  207. cgrpc_metadata_array *metadata = cgrpc_metadata_array_create();
  208. cgrpc_metadata_array_move_metadata(metadata, &(observer->trailing_metadata_recv));
  209. return metadata;
  210. }
  211. long cgrpc_observer_recv_status_on_client_get_status(cgrpc_observer_recv_status_on_client *observer) {
  212. return observer->server_status;
  213. }
  214. char *cgrpc_observer_recv_status_on_client_copy_status_details(cgrpc_observer_recv_status_on_client *observer) {
  215. size_t length = GRPC_SLICE_LENGTH(observer->server_details);
  216. char *str = (char *) malloc(length + 1);
  217. memcpy(str, GRPC_SLICE_START_PTR(observer->server_details), length);
  218. str[length] = 0;
  219. return str;
  220. }
  221. int cgrpc_observer_recv_close_on_server_get_was_cancelled(cgrpc_observer_recv_close_on_server *observer) {
  222. return observer->was_cancelled;
  223. }