// Copyright 2007 - 2021, Alan Antonuk and the rabbitmq-c contributors. // SPDX-License-Identifier: mit #ifndef librabbitmq_amqp_private_h #define librabbitmq_amqp_private_h #ifdef HAVE_CONFIG_H #include "config.h" #endif #define AMQ_COPYRIGHT \ "Copyright (c) 2007-2014 VMWare Inc, Tony Garnock-Jones," \ " and Alan Antonuk." #include "rabbitmq-c/amqp.h" #include "rabbitmq-c/framing.h" #include #if ((defined(_WIN32)) || (defined(__MINGW32__)) || (defined(__MINGW64__))) #ifndef WINVER /* WINVER 0x0502 is WinXP SP2+, Windows Server 2003 SP1+ * See: * http://msdn.microsoft.com/en-us/library/windows/desktop/aa383745(v=vs.85).aspx#macros_for_conditional_declarations */ #define WINVER 0x0502 #endif #ifndef WIN32_LEAN_AND_MEAN #define WIN32_LEAN_AND_MEAN #endif #include #else #include #include #endif /* GCC attributes */ #if __GNUC__ > 2 || (__GNUC__ == 2 && __GNUC_MINOR__ > 4) #define AMQP_NORETURN __attribute__((__noreturn__)) #define AMQP_UNUSED __attribute__((__unused__)) #elif defined(_MSC_VER) #define AMQP_NORETURN __declspec(noreturn) #define AMQP_UNUSED __pragma(warning(suppress : 4100)) #else #define AMQP_NORETURN #define AMQP_UNUSED #endif #if (defined(_MSC_VER) && (_MSC_VER <= 1800)) || \ (defined(__BORLANDC__) && (__BORLANDC__ <= 0x0564)) #define inline __inline #endif char *amqp_os_error_string(int err); #include "amqp_socket.h" #include "amqp_time.h" /* * Connection states: XXX FIX THIS * * - CONNECTION_STATE_INITIAL: The initial state, when we cannot be * sure if the next thing we will get is the first AMQP frame, or a * protocol header from the server. * * - CONNECTION_STATE_IDLE: The normal state between * frames. Connections may only be reconfigured, and the * connection's pools recycled, when in this state. Whenever we're * in this state, the inbound_buffer's bytes pointer must be NULL; * any other state, and it must point to a block of memory allocated * from the frame_pool. * * - CONNECTION_STATE_HEADER: Some bytes of an incoming frame have * been seen, but not a complete frame header's worth. * * - CONNECTION_STATE_BODY: A complete frame header has been seen, but * the frame is not yet complete. When it is completed, it will be * returned, and the connection will return to IDLE state. * */ typedef enum amqp_connection_state_enum_ { CONNECTION_STATE_IDLE = 0, CONNECTION_STATE_INITIAL, CONNECTION_STATE_HEADER, CONNECTION_STATE_BODY } amqp_connection_state_enum; typedef enum amqp_status_private_enum_ { /* 0x00xx -> AMQP_STATUS_*/ /* 0x01xx -> AMQP_STATUS_TCP_* */ /* 0x02xx -> AMQP_STATUS_SSL_* */ AMQP_PRIVATE_STATUS_SOCKET_NEEDREAD = -0x1301, AMQP_PRIVATE_STATUS_SOCKET_NEEDWRITE = -0x1302 } amqp_status_private_enum; /* 7 bytes up front, then payload, then 1 byte footer */ #define HEADER_SIZE 7 #define FOOTER_SIZE 1 #define AMQP_PSEUDOFRAME_PROTOCOL_HEADER 'A' typedef struct amqp_link_t_ { struct amqp_link_t_ *next; void *data; } amqp_link_t; #define POOL_TABLE_SIZE 16 typedef struct amqp_pool_table_entry_t_ { struct amqp_pool_table_entry_t_ *next; amqp_pool_t pool; amqp_channel_t channel; } amqp_pool_table_entry_t; struct amqp_connection_state_t_ { amqp_pool_table_entry_t *pool_table[POOL_TABLE_SIZE]; amqp_connection_state_enum state; int channel_max; int frame_max; /* Heartbeat interval in seconds. If this is <= 0, then heartbeats are not * enabled, and next_recv_heartbeat and next_send_heartbeat are set to * infinite */ int heartbeat; amqp_time_t next_recv_heartbeat; amqp_time_t next_send_heartbeat; /* buffer for holding frame headers. Allows us to delay allocating * the raw frame buffer until the type, channel, and size are all known */ char header_buffer[HEADER_SIZE + 1]; amqp_bytes_t inbound_buffer; size_t inbound_offset; size_t target_size; amqp_bytes_t outbound_buffer; amqp_socket_t *socket; amqp_bytes_t sock_inbound_buffer; size_t sock_inbound_offset; size_t sock_inbound_limit; amqp_link_t *first_queued_frame; amqp_link_t *last_queued_frame; amqp_rpc_reply_t most_recent_api_result; amqp_table_t server_properties; amqp_table_t client_properties; amqp_pool_t properties_pool; struct timeval *handshake_timeout; struct timeval internal_handshake_timeout; struct timeval *rpc_timeout; struct timeval internal_rpc_timeout; }; amqp_pool_t *amqp_get_or_create_channel_pool(amqp_connection_state_t connection, amqp_channel_t channel); amqp_pool_t *amqp_get_channel_pool(amqp_connection_state_t state, amqp_channel_t channel); static inline int amqp_heartbeat_send(amqp_connection_state_t state) { return state->heartbeat; } static inline int amqp_heartbeat_recv(amqp_connection_state_t state) { return 2 * state->heartbeat; } int amqp_try_recv(amqp_connection_state_t state); static inline void *amqp_offset(void *data, size_t offset) { return (char *)data + offset; } /* This macro defines the encoding and decoding functions associated with a simple type. */ #define DECLARE_CODEC_BASE_TYPE(bits) \ \ static inline int amqp_encode_##bits(amqp_bytes_t encoded, size_t *offset, \ uint##bits##_t input) { \ size_t o = *offset; \ if ((*offset = o + bits / 8) <= encoded.len) { \ amqp_e##bits(input, amqp_offset(encoded.bytes, o)); \ return 1; \ } \ return 0; \ } \ \ static inline int amqp_decode_##bits(amqp_bytes_t encoded, size_t *offset, \ uint##bits##_t *output) { \ size_t o = *offset; \ if ((*offset = o + bits / 8) <= encoded.len) { \ *output = amqp_d##bits(amqp_offset(encoded.bytes, o)); \ return 1; \ } \ return 0; \ } static inline int is_bigendian(void) { union { uint32_t i; char c[4]; } bint = {0x01020304}; return bint.c[0] == 1; } static inline void amqp_e8(uint8_t val, void *data) { memcpy(data, &val, sizeof(val)); } static inline uint8_t amqp_d8(void *data) { uint8_t val; memcpy(&val, data, sizeof(val)); return val; } static inline void amqp_e16(uint16_t val, void *data) { if (!is_bigendian()) { val = ((val & 0xFF00u) >> 8u) | ((val & 0x00FFu) << 8u); } memcpy(data, &val, sizeof(val)); } static inline uint16_t amqp_d16(void *data) { uint16_t val; memcpy(&val, data, sizeof(val)); if (!is_bigendian()) { val = ((val & 0xFF00u) >> 8u) | ((val & 0x00FFu) << 8u); } return val; } static inline void amqp_e32(uint32_t val, void *data) { if (!is_bigendian()) { val = ((val & 0xFF000000u) >> 24u) | ((val & 0x00FF0000u) >> 8u) | ((val & 0x0000FF00u) << 8u) | ((val & 0x000000FFu) << 24u); } memcpy(data, &val, sizeof(val)); } static inline uint32_t amqp_d32(void *data) { uint32_t val; memcpy(&val, data, sizeof(val)); if (!is_bigendian()) { val = ((val & 0xFF000000u) >> 24u) | ((val & 0x00FF0000u) >> 8u) | ((val & 0x0000FF00u) << 8u) | ((val & 0x000000FFu) << 24u); } return val; } static inline void amqp_e64(uint64_t val, void *data) { if (!is_bigendian()) { val = ((val & 0xFF00000000000000u) >> 56u) | ((val & 0x00FF000000000000u) >> 40u) | ((val & 0x0000FF0000000000u) >> 24u) | ((val & 0x000000FF00000000u) >> 8u) | ((val & 0x00000000FF000000u) << 8u) | ((val & 0x0000000000FF0000u) << 24u) | ((val & 0x000000000000FF00u) << 40u) | ((val & 0x00000000000000FFu) << 56u); } memcpy(data, &val, sizeof(val)); } static inline uint64_t amqp_d64(void *data) { uint64_t val; memcpy(&val, data, sizeof(val)); if (!is_bigendian()) { val = ((val & 0xFF00000000000000u) >> 56u) | ((val & 0x00FF000000000000u) >> 40u) | ((val & 0x0000FF0000000000u) >> 24u) | ((val & 0x000000FF00000000u) >> 8u) | ((val & 0x00000000FF000000u) << 8u) | ((val & 0x0000000000FF0000u) << 24u) | ((val & 0x000000000000FF00u) << 40u) | ((val & 0x00000000000000FFu) << 56u); } return val; } DECLARE_CODEC_BASE_TYPE(8) DECLARE_CODEC_BASE_TYPE(16) DECLARE_CODEC_BASE_TYPE(32) DECLARE_CODEC_BASE_TYPE(64) static inline int amqp_encode_bytes(amqp_bytes_t encoded, size_t *offset, amqp_bytes_t input) { size_t o = *offset; /* The memcpy below has undefined behavior if the input is NULL. It is valid * for a 0-length amqp_bytes_t to have .bytes == NULL. Thus we should check * before encoding. */ if (input.len == 0) { return 1; } if ((*offset = o + input.len) <= encoded.len) { memcpy(amqp_offset(encoded.bytes, o), input.bytes, input.len); return 1; } else { return 0; } } static inline int amqp_decode_bytes(amqp_bytes_t encoded, size_t *offset, amqp_bytes_t *output, size_t len) { size_t o = *offset; if ((*offset = o + len) <= encoded.len) { output->bytes = amqp_offset(encoded.bytes, o); output->len = len; return 1; } else { return 0; } } AMQP_NORETURN void amqp_abort(const char *fmt, ...); int amqp_bytes_equal(amqp_bytes_t r, amqp_bytes_t l); static inline amqp_rpc_reply_t amqp_rpc_reply_error(amqp_status_enum status) { amqp_rpc_reply_t reply; reply.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION; reply.library_error = status; return reply; } int amqp_send_frame_inner(amqp_connection_state_t state, const amqp_frame_t *frame, int flags, amqp_time_t deadline); #endif