/src/librabbitmq/librabbitmq/amqp_consumer.c

Source
// Copyright 2007 - 2021, Alan Antonuk and the rabbitmq-c contributors.
// SPDX-License-Identifier: mit

#include "amqp_private.h"
#include "amqp_socket.h"
#include "rabbitmq-c/amqp.h"

#include <stdlib.h>
#include <string.h>

static int amqp_basic_properties_clone(amqp_basic_properties_t *original,
                                       amqp_basic_properties_t *clone,
                                       amqp_pool_t *pool) {
  memset(clone, 0, sizeof(*clone));
  clone->_flags = original->_flags;

#define CLONE_BYTES_POOL(original, clone, pool)        \
  if (0 == original.len) {                             \
    clone = amqp_empty_bytes;                          \
  } else {                                             \
    amqp_pool_alloc_bytes(pool, original.len, &clone); \
    if (NULL == clone.bytes) {                         \
      return AMQP_STATUS_NO_MEMORY;                    \
    }                                                  \
    memcpy(clone.bytes, original.bytes, clone.len);    \
  }

  if (clone->_flags & AMQP_BASIC_CONTENT_TYPE_FLAG) {
    CLONE_BYTES_POOL(original->content_type, clone->content_type, pool)
  }

  if (clone->_flags & AMQP_BASIC_CONTENT_ENCODING_FLAG) {
    CLONE_BYTES_POOL(original->content_encoding, clone->content_encoding, pool)
  }

  if (clone->_flags & AMQP_BASIC_HEADERS_FLAG) {
    int res = amqp_table_clone(&original->headers, &clone->headers, pool);
    if (AMQP_STATUS_OK != res) {
      return res;
    }
  }

  if (clone->_flags & AMQP_BASIC_DELIVERY_MODE_FLAG) {
    clone->delivery_mode = original->delivery_mode;
  }

  if (clone->_flags & AMQP_BASIC_PRIORITY_FLAG) {
    clone->priority = original->priority;
  }

  if (clone->_flags & AMQP_BASIC_CORRELATION_ID_FLAG) {
    CLONE_BYTES_POOL(original->correlation_id, clone->correlation_id, pool)
  }

  if (clone->_flags & AMQP_BASIC_REPLY_TO_FLAG) {
    CLONE_BYTES_POOL(original->reply_to, clone->reply_to, pool)
  }

  if (clone->_flags & AMQP_BASIC_EXPIRATION_FLAG) {
    CLONE_BYTES_POOL(original->expiration, clone->expiration, pool)
  }

  if (clone->_flags & AMQP_BASIC_MESSAGE_ID_FLAG) {
    CLONE_BYTES_POOL(original->message_id, clone->message_id, pool)
  }

  if (clone->_flags & AMQP_BASIC_TIMESTAMP_FLAG) {
    clone->timestamp = original->timestamp;
  }

  if (clone->_flags & AMQP_BASIC_TYPE_FLAG) {
    CLONE_BYTES_POOL(original->type, clone->type, pool)
  }

  if (clone->_flags & AMQP_BASIC_USER_ID_FLAG) {
    CLONE_BYTES_POOL(original->user_id, clone->user_id, pool)
  }

  if (clone->_flags & AMQP_BASIC_APP_ID_FLAG) {
    CLONE_BYTES_POOL(original->app_id, clone->app_id, pool)
  }

  if (clone->_flags & AMQP_BASIC_CLUSTER_ID_FLAG) {
    CLONE_BYTES_POOL(original->cluster_id, clone->cluster_id, pool)
  }

  return AMQP_STATUS_OK;
#undef CLONE_BYTES_POOL
}

void amqp_destroy_message(amqp_message_t *message) {
  empty_amqp_pool(&message->pool);
  amqp_bytes_free(message->body);
}

void amqp_destroy_envelope(amqp_envelope_t *envelope) {
  amqp_destroy_message(&envelope->message);
  amqp_bytes_free(envelope->routing_key);
  amqp_bytes_free(envelope->exchange);
  amqp_bytes_free(envelope->consumer_tag);
}

static int amqp_bytes_malloc_dup_failed(amqp_bytes_t bytes) {
  if (bytes.len != 0 && bytes.bytes == NULL) {
    return 1;
  }
  return 0;
}

amqp_rpc_reply_t amqp_consume_message(amqp_connection_state_t state,
                                      amqp_envelope_t *envelope,
                                      const struct timeval *timeout,
                                      AMQP_UNUSED int flags) {
  int res;
  amqp_frame_t frame;
  amqp_basic_deliver_t *delivery_method;
  amqp_rpc_reply_t ret;

  memset(&ret, 0, sizeof(ret));
  memset(envelope, 0, sizeof(*envelope));

  res = amqp_simple_wait_frame_noblock(state, &frame, timeout);
  if (AMQP_STATUS_OK != res) {
    ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
    ret.library_error = res;
    goto error_out1;
  }

  if (AMQP_FRAME_METHOD != frame.frame_type ||
      AMQP_BASIC_DELIVER_METHOD != frame.payload.method.id) {
    amqp_put_back_frame(state, &frame);
    ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
    ret.library_error = AMQP_STATUS_UNEXPECTED_STATE;
    goto error_out1;
  }

  delivery_method = frame.payload.method.decoded;

  envelope->channel = frame.channel;
  envelope->consumer_tag = amqp_bytes_malloc_dup(delivery_method->consumer_tag);
  envelope->delivery_tag = delivery_method->delivery_tag;
  envelope->redelivered = delivery_method->redelivered;
  envelope->exchange = amqp_bytes_malloc_dup(delivery_method->exchange);
  envelope->routing_key = amqp_bytes_malloc_dup(delivery_method->routing_key);

  if (amqp_bytes_malloc_dup_failed(envelope->consumer_tag) ||
      amqp_bytes_malloc_dup_failed(envelope->exchange) ||
      amqp_bytes_malloc_dup_failed(envelope->routing_key)) {
    ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
    ret.library_error = AMQP_STATUS_NO_MEMORY;
    goto error_out2;
  }

  ret = amqp_read_message(state, envelope->channel, &envelope->message, 0);
  if (AMQP_RESPONSE_NORMAL != ret.reply_type) {
    goto error_out2;
  }

  ret.reply_type = AMQP_RESPONSE_NORMAL;
  return ret;

error_out2:
  amqp_bytes_free(envelope->routing_key);
  amqp_bytes_free(envelope->exchange);
  amqp_bytes_free(envelope->consumer_tag);
error_out1:
  return ret;
}

amqp_rpc_reply_t amqp_read_message(amqp_connection_state_t state,
                                   amqp_channel_t channel,
                                   amqp_message_t *message,
                                   AMQP_UNUSED int flags) {
  amqp_frame_t frame;
  amqp_rpc_reply_t ret;

  size_t body_read;
  char *body_read_ptr;
  int res;

  memset(&ret, 0, sizeof(ret));
  memset(message, 0, sizeof(*message));

  res = amqp_simple_wait_frame_on_channel(state, channel, &frame);
  if (AMQP_STATUS_OK != res) {
    ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
    ret.library_error = res;

    goto error_out1;
  }

  if (AMQP_FRAME_HEADER != frame.frame_type) {
    if (AMQP_FRAME_METHOD == frame.frame_type &&
        (AMQP_CHANNEL_CLOSE_METHOD == frame.payload.method.id ||
         AMQP_CONNECTION_CLOSE_METHOD == frame.payload.method.id)) {

      ret.reply_type = AMQP_RESPONSE_SERVER_EXCEPTION;
      ret.reply = frame.payload.method;

    } else {
      ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
      ret.library_error = AMQP_STATUS_UNEXPECTED_STATE;

      amqp_put_back_frame(state, &frame);
    }
    goto error_out1;
  }

  if (AMQP_BASIC_CLASS != frame.payload.properties.class_id) {
    ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
    ret.library_error = AMQP_STATUS_BAD_AMQP_DATA;
    goto error_out1;
  }

  init_amqp_pool(&message->pool, 4096);
  res = amqp_basic_properties_clone(frame.payload.properties.decoded,
                                    &message->properties, &message->pool);

  if (AMQP_STATUS_OK != res) {
    ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
    ret.library_error = res;
    goto error_out3;
  }

  if (0 == frame.payload.properties.body_size) {
    message->body = amqp_empty_bytes;
  } else {
    if (SIZE_MAX < frame.payload.properties.body_size) {
      ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
      ret.library_error = AMQP_STATUS_NO_MEMORY;
      goto error_out1;
    }
    message->body =
        amqp_bytes_malloc((size_t)frame.payload.properties.body_size);
    if (NULL == message->body.bytes) {
      ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
      ret.library_error = AMQP_STATUS_NO_MEMORY;
      goto error_out1;
    }
  }

  body_read = 0;
  body_read_ptr = message->body.bytes;

  while (body_read < message->body.len) {
    res = amqp_simple_wait_frame_on_channel(state, channel, &frame);
    if (AMQP_STATUS_OK != res) {
      ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
      ret.library_error = res;
      goto error_out2;
    }
    if (AMQP_FRAME_BODY != frame.frame_type) {
      if (AMQP_FRAME_METHOD == frame.frame_type &&
          (AMQP_CHANNEL_CLOSE_METHOD == frame.payload.method.id ||
           AMQP_CONNECTION_CLOSE_METHOD == frame.payload.method.id)) {

        ret.reply_type = AMQP_RESPONSE_SERVER_EXCEPTION;
        ret.reply = frame.payload.method;
      } else {
        ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
        ret.library_error = AMQP_STATUS_BAD_AMQP_DATA;
      }
      goto error_out2;
    }

    if (body_read + frame.payload.body_fragment.len > message->body.len) {
      ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
      ret.library_error = AMQP_STATUS_BAD_AMQP_DATA;
      goto error_out2;
    }

    memcpy(body_read_ptr, frame.payload.body_fragment.bytes,
           frame.payload.body_fragment.len);

    body_read += frame.payload.body_fragment.len;
    body_read_ptr += frame.payload.body_fragment.len;
  }

  ret.reply_type = AMQP_RESPONSE_NORMAL;
  return ret;

error_out2:
  amqp_bytes_free(message->body);
error_out3:
  empty_amqp_pool(&message->pool);
error_out1:
  return ret;
}

Coverage Report

Created: 2026-10-28 06:06

Line	Count	Source
1		// Copyright 2007 - 2021, Alan Antonuk and the rabbitmq-c contributors.
2		// SPDX-License-Identifier: mit
3
4		#include "amqp_private.h"
5		#include "amqp_socket.h"
6		#include "rabbitmq-c/amqp.h"
7
8		#include <stdlib.h>
9		#include <string.h>
10
11		static int amqp_basic_properties_clone(amqp_basic_properties_t *original,
12		amqp_basic_properties_t *clone,
13	546	amqp_pool_t *pool) {
14	546	memset(clone, 0, sizeof(*clone));
15	546	clone->_flags = original->_flags;
16
17	546	#define CLONE_BYTES_POOL(original, clone, pool) \
18	1.03k	if (0 == original.len) { \
19	518	clone = amqp_empty_bytes; \
20	521	} else { \
21	521	amqp_pool_alloc_bytes(pool, original.len, &clone); \
22	521	if (NULL == clone.bytes) { \
23	0	return AMQP_STATUS_NO_MEMORY; \
24	0	} \
25	521	memcpy(clone.bytes, original.bytes, clone.len); \
26	521	}
27
28	546	if (clone->_flags & AMQP_BASIC_CONTENT_TYPE_FLAG) {
29	40	CLONE_BYTES_POOL(original->content_type, clone->content_type, pool)
30	40	}
31
32	546	if (clone->_flags & AMQP_BASIC_CONTENT_ENCODING_FLAG) {
33	160	CLONE_BYTES_POOL(original->content_encoding, clone->content_encoding, pool)
34	160	}
35
36	546	if (clone->_flags & AMQP_BASIC_HEADERS_FLAG) {
37	301	int res = amqp_table_clone(&original->headers, &clone->headers, pool);
38	301	if (AMQP_STATUS_OK != res) {
39	27	return res;
40	27	}
41	301	}
42
43	519	if (clone->_flags & AMQP_BASIC_DELIVERY_MODE_FLAG) {
44	87	clone->delivery_mode = original->delivery_mode;
45	87	}
46
47	519	if (clone->_flags & AMQP_BASIC_PRIORITY_FLAG) {
48	69	clone->priority = original->priority;
49	69	}
50
51	519	if (clone->_flags & AMQP_BASIC_CORRELATION_ID_FLAG) {
52	191	CLONE_BYTES_POOL(original->correlation_id, clone->correlation_id, pool)
53	191	}
54
55	519	if (clone->_flags & AMQP_BASIC_REPLY_TO_FLAG) {
56	100	CLONE_BYTES_POOL(original->reply_to, clone->reply_to, pool)
57	100	}
58
59	519	if (clone->_flags & AMQP_BASIC_EXPIRATION_FLAG) {
60	143	CLONE_BYTES_POOL(original->expiration, clone->expiration, pool)
61	143	}
62
63	519	if (clone->_flags & AMQP_BASIC_MESSAGE_ID_FLAG) {
64	86	CLONE_BYTES_POOL(original->message_id, clone->message_id, pool)
65	86	}
66
67	519	if (clone->_flags & AMQP_BASIC_TIMESTAMP_FLAG) {
68	60	clone->timestamp = original->timestamp;
69	60	}
70
71	519	if (clone->_flags & AMQP_BASIC_TYPE_FLAG) {
72	55	CLONE_BYTES_POOL(original->type, clone->type, pool)
73	55	}
74
75	519	if (clone->_flags & AMQP_BASIC_USER_ID_FLAG) {
76	87	CLONE_BYTES_POOL(original->user_id, clone->user_id, pool)
77	87	}
78
79	519	if (clone->_flags & AMQP_BASIC_APP_ID_FLAG) {
80	86	CLONE_BYTES_POOL(original->app_id, clone->app_id, pool)
81	86	}
82
83	519	if (clone->_flags & AMQP_BASIC_CLUSTER_ID_FLAG) {
84	91	CLONE_BYTES_POOL(original->cluster_id, clone->cluster_id, pool)
85	91	}
86
87	519	return AMQP_STATUS_OK;
88	519	#undef CLONE_BYTES_POOL
89	519	}
90
91	15	void amqp_destroy_message(amqp_message_t *message) {
92	15	empty_amqp_pool(&message->pool);
93	15	amqp_bytes_free(message->body);
94	15	}
95
96	0	void amqp_destroy_envelope(amqp_envelope_t *envelope) {
97	0	amqp_destroy_message(&envelope->message);
98	0	amqp_bytes_free(envelope->routing_key);
99	0	amqp_bytes_free(envelope->exchange);
100	0	amqp_bytes_free(envelope->consumer_tag);
101	0	}
102
103	0	static int amqp_bytes_malloc_dup_failed(amqp_bytes_t bytes) {
104	0	if (bytes.len != 0 && bytes.bytes == NULL) {
105	0	return 1;
106	0	}
107	0	return 0;
108	0	}
109
110		amqp_rpc_reply_t amqp_consume_message(amqp_connection_state_t state,
111		amqp_envelope_t *envelope,
112		const struct timeval *timeout,
113	0	AMQP_UNUSED int flags) {
114	0	int res;
115	0	amqp_frame_t frame;
116	0	amqp_basic_deliver_t *delivery_method;
117	0	amqp_rpc_reply_t ret;
118
119	0	memset(&ret, 0, sizeof(ret));
120	0	memset(envelope, 0, sizeof(*envelope));
121
122	0	res = amqp_simple_wait_frame_noblock(state, &frame, timeout);
123	0	if (AMQP_STATUS_OK != res) {
124	0	ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
125	0	ret.library_error = res;
126	0	goto error_out1;
127	0	}
128
129	0	if (AMQP_FRAME_METHOD != frame.frame_type \|\|
130	0	AMQP_BASIC_DELIVER_METHOD != frame.payload.method.id) {
131	0	amqp_put_back_frame(state, &frame);
132	0	ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
133	0	ret.library_error = AMQP_STATUS_UNEXPECTED_STATE;
134	0	goto error_out1;
135	0	}
136
137	0	delivery_method = frame.payload.method.decoded;
138
139	0	envelope->channel = frame.channel;
140	0	envelope->consumer_tag = amqp_bytes_malloc_dup(delivery_method->consumer_tag);
141	0	envelope->delivery_tag = delivery_method->delivery_tag;
142	0	envelope->redelivered = delivery_method->redelivered;
143	0	envelope->exchange = amqp_bytes_malloc_dup(delivery_method->exchange);
144	0	envelope->routing_key = amqp_bytes_malloc_dup(delivery_method->routing_key);
145
146	0	if (amqp_bytes_malloc_dup_failed(envelope->consumer_tag) \|\|
147	0	amqp_bytes_malloc_dup_failed(envelope->exchange) \|\|
148	0	amqp_bytes_malloc_dup_failed(envelope->routing_key)) {
149	0	ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
150	0	ret.library_error = AMQP_STATUS_NO_MEMORY;
151	0	goto error_out2;
152	0	}
153
154	0	ret = amqp_read_message(state, envelope->channel, &envelope->message, 0);
155	0	if (AMQP_RESPONSE_NORMAL != ret.reply_type) {
156	0	goto error_out2;
157	0	}
158
159	0	ret.reply_type = AMQP_RESPONSE_NORMAL;
160	0	return ret;
161
162	0	error_out2:
163	0	amqp_bytes_free(envelope->routing_key);
164	0	amqp_bytes_free(envelope->exchange);
165	0	amqp_bytes_free(envelope->consumer_tag);
166	0	error_out1:
167	0	return ret;
168	0	}
169
170		amqp_rpc_reply_t amqp_read_message(amqp_connection_state_t state,
171		amqp_channel_t channel,
172		amqp_message_t *message,
173	5.47k	AMQP_UNUSED int flags) {
174	5.47k	amqp_frame_t frame;
175	5.47k	amqp_rpc_reply_t ret;
176
177	5.47k	size_t body_read;
178	5.47k	char *body_read_ptr;
179	5.47k	int res;
180
181	5.47k	memset(&ret, 0, sizeof(ret));
182	5.47k	memset(message, 0, sizeof(*message));
183
184	5.47k	res = amqp_simple_wait_frame_on_channel(state, channel, &frame);
185	5.47k	if (AMQP_STATUS_OK != res) {
186	4.87k	ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
187	4.87k	ret.library_error = res;
188
189	4.87k	goto error_out1;
190	4.87k	}
191
192	593	if (AMQP_FRAME_HEADER != frame.frame_type) {
193	39	if (AMQP_FRAME_METHOD == frame.frame_type &&
194	35	(AMQP_CHANNEL_CLOSE_METHOD == frame.payload.method.id \|\|
195	34	AMQP_CONNECTION_CLOSE_METHOD == frame.payload.method.id)) {
196
197	3	ret.reply_type = AMQP_RESPONSE_SERVER_EXCEPTION;
198	3	ret.reply = frame.payload.method;
199
200	36	} else {
201	36	ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
202	36	ret.library_error = AMQP_STATUS_UNEXPECTED_STATE;
203
204	36	amqp_put_back_frame(state, &frame);
205	36	}
206	39	goto error_out1;
207	39	}
208
209	554	if (AMQP_BASIC_CLASS != frame.payload.properties.class_id) {
210	8	ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
211	8	ret.library_error = AMQP_STATUS_BAD_AMQP_DATA;
212	8	goto error_out1;
213	8	}
214
215	546	init_amqp_pool(&message->pool, 4096);
216	546	res = amqp_basic_properties_clone(frame.payload.properties.decoded,
217	546	&message->properties, &message->pool);
218
219	546	if (AMQP_STATUS_OK != res) {
220	27	ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
221	27	ret.library_error = res;
222	27	goto error_out3;
223	27	}
224
225	519	if (0 == frame.payload.properties.body_size) {
226	14	message->body = amqp_empty_bytes;
227	505	} else {
228	505	if (SIZE_MAX < frame.payload.properties.body_size) {
229	0	ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
230	0	ret.library_error = AMQP_STATUS_NO_MEMORY;
231	0	goto error_out1;
232	0	}
233	505	message->body =
234	505	amqp_bytes_malloc((size_t)frame.payload.properties.body_size);
235	505	if (NULL == message->body.bytes) {
236	0	ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
237	0	ret.library_error = AMQP_STATUS_NO_MEMORY;
238	0	goto error_out1;
239	0	}
240	505	}
241
242	519	body_read = 0;
243	519	body_read_ptr = message->body.bytes;
244
245	1.11k	while (body_read < message->body.len) {
246	1.10k	res = amqp_simple_wait_frame_on_channel(state, channel, &frame);
247	1.10k	if (AMQP_STATUS_OK != res) {
248	467	ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
249	467	ret.library_error = res;
250	467	goto error_out2;
251	467	}
252	633	if (AMQP_FRAME_BODY != frame.frame_type) {
253	21	if (AMQP_FRAME_METHOD == frame.frame_type &&
254	16	(AMQP_CHANNEL_CLOSE_METHOD == frame.payload.method.id \|\|
255	15	AMQP_CONNECTION_CLOSE_METHOD == frame.payload.method.id)) {
256
257	2	ret.reply_type = AMQP_RESPONSE_SERVER_EXCEPTION;
258	2	ret.reply = frame.payload.method;
259	19	} else {
260	19	ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
261	19	ret.library_error = AMQP_STATUS_BAD_AMQP_DATA;
262	19	}
263	21	goto error_out2;
264	21	}
265
266	612	if (body_read + frame.payload.body_fragment.len > message->body.len) {
267	16	ret.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
268	16	ret.library_error = AMQP_STATUS_BAD_AMQP_DATA;
269	16	goto error_out2;
270	16	}
271
272	596	memcpy(body_read_ptr, frame.payload.body_fragment.bytes,
273	596	frame.payload.body_fragment.len);
274
275	596	body_read += frame.payload.body_fragment.len;
276	596	body_read_ptr += frame.payload.body_fragment.len;
277	596	}
278
279	15	ret.reply_type = AMQP_RESPONSE_NORMAL;
280	15	return ret;
281
282	504	error_out2:
283	504	amqp_bytes_free(message->body);
284	531	error_out3:
285	531	empty_amqp_pool(&message->pool);
286	5.45k	error_out1:
287	5.45k	return ret;
288	531	}