Return to pa_vamqp.h CVS log

Up to [parser3project] / parser3 / src / types

Annotation of parser3/src/types/pa_vamqp.h, revision 1.6

1.1       moko        1: /** @file
                      2:        Parser: @b amqp class decls.
                      3: 
1.2       moko        4:        Copyright (c) 2001-2026 Art. Lebedev Studio (https://www.artlebedev.com)
1.1       moko        5:        Authors: Konstantin Morshnev <moko@design.ru>
                      6: */
                      7: 
                      8: #ifndef PA_VAMQP_H
                      9: #define PA_VAMQP_H
                     10: 
1.6     ! moko       11: #define IDENT_PA_VAMQP_H "$Id: pa_vamqp.h,v 1.5 2026/10/09 18:32:15 moko Exp $"
1.1       moko       12: 
                     13: #include "classes.h"
                     14: #include "pa_vstateless_object.h"
1.3       moko       15: #include "pa_common.h"
1.6     ! moko       16: #include "pa_pool.h"
1.1       moko       17: 
1.5       moko       18: //for librabbitmq before 0.12
                     19: //#define PA_AMQP_COMPAT
                     20: 
1.1       moko       21: #ifdef WITH_AMQP
1.5       moko       22: 
                     23: #ifdef PA_AMQP_COMPAT
                     24: 
1.1       moko       25: #include <amqp.h>
1.5       moko       26: #include <amqp_tcp_socket.h>
                     27: #include <amqp_framing.h>
                     28: #ifdef WITH_AMQP_SSL
                     29: #include <amqp_ssl_socket.h>
                     30: #endif
                     31: 
                     32: #else
                     33: 
                     34: #include <rabbitmq-c/amqp.h>
                     35: #include <rabbitmq-c/tcp_socket.h>
                     36: #include <rabbitmq-c/framing.h>
                     37: #ifdef WITH_AMQP_SSL
                     38: #include <rabbitmq-c/ssl_socket.h>
                     39: #endif
                     40: 
                     41: #endif
                     42: 
1.1       moko       43: #endif
                     44: 
                     45: // defines
                     46: #define VAMQP_TYPE "amqp"
                     47: 
                     48: // externs
                     49: extern Methoded *amqp_class;
                     50: 
1.6     ! moko       51: class VAmqp: public VStateless_object, Pooled {
1.1       moko       52: public:
                     53:        // value
                     54:        override const char* type() const { return VAMQP_TYPE; }
                     55:        override VStateless_class *get_class() { return amqp_class; }
                     56: 
                     57: #ifdef WITH_AMQP
                     58: public: // usage
1.3       moko       59: 
                     60:        // ALIVE: usable right now
                     61:        // CHANNEL_DEAD: last operation got a channel-level error - reopening the channel (same connection) is required
                     62:        // CONNECTION_DEAD: last operation got a connection-level error - a full reconnect is required
                     63:        enum ConnState { ALIVE, CHANNEL_DEAD, CONNECTION_DEAD };
                     64: 
1.6     ! moko       65:        VAmqp(Pool& apool): Pooled(apool), fconnection(0), fchannel(0), fstate(CONNECTION_DEAD), fcreate_options(0), freconnect_interval(0) {}
        !            66: 
        !            67:        /// called by the request pool at the end of the request
        !            68:        override ~VAmqp() { release(); }
1.1       moko       69: 
                     70:        amqp_connection_state_t fconnection;
1.4       moko       71:        amqp_channel_t fchannel;
1.1       moko       72:        bool fstop;
1.3       moko       73:        ConnState fstate;
                     74:        HashStringValue* fcreate_options;
                     75:        int freconnect_interval; // 0 = auto-reconnect disabled, N>0 = enabled, sleeps N sec before each attempt
                     76: 
                     77:        void ensure_connected();
1.1       moko       78: 
                     79:        amqp_connection_state_t connection() {
                     80:                if(!fconnection)
                     81:                        throw Exception(PARSER_RUNTIME, 0, "using uninitialized amqp object");
1.3       moko       82:                ensure_connected();
1.1       moko       83:                return fconnection;
                     84:        }
                     85: 
1.4       moko       86:        amqp_channel_t channel() {
1.1       moko       87:                if(!fchannel)
                     88:                        throw Exception(PARSER_RUNTIME, 0, "using uninitialized amqp object channel");
1.3       moko       89:                ensure_connected();
1.1       moko       90:                return fchannel;
                     91:        }
                     92: 
1.6     ! moko       93:        void release() {
        !            94:                if(!fconnection)
        !            95:                        return;
        !            96:                if(fstate==ALIVE) { // a dead connection or channel is not closed politely
        !            97:                        amqp_channel_close(fconnection, fchannel, AMQP_REPLY_SUCCESS);
        !            98:                        amqp_connection_close(fconnection, AMQP_REPLY_SUCCESS);
        !            99:                }
        !           100:                amqp_destroy_connection(fconnection);
        !           101:                fconnection=0;
        !           102:                fchannel=0;
        !           103:                fstate=CONNECTION_DEAD;
        !           104:        }
        !           105: 
        !           106: #else
        !           107:        VAmqp(Pool& apool): Pooled(apool) {}
1.1       moko      108: #endif // WITH_AMQP
                    109: };
                    110: 
                    111: #endif
                    112: 
                    113: