[BACK]Return to amqp.C CVS log [TXT] [DIR] Up to [parser3project] / parser3 / src / classes

Diff for /parser3/src/classes/amqp.C between versions 1.1 and 1.4

version 1.1, 2025/11/06 22:08:03 version 1.4, 2025/11/07 23:00:11
Line 101  VAmqp& self=GET_SELF(r, VAmqp); Line 101  VAmqp& self=GET_SELF(r, VAmqp);
   
 #ifdef WITH_AMQP  #ifdef WITH_AMQP
   
 static void check(const char* action, amqp_rpc_reply_t rr){  static void check(amqp_rpc_reply_t rr){
         if(rr.reply_type != AMQP_RESPONSE_NORMAL)          if(rr.reply_type == AMQP_RESPONSE_NORMAL)
                 throw Exception("amqp", 0, "%s failed", action);                  return;
           
           // Extract error message from reply
           const char* error_msg = 0;
           size_t error_len = 0;
           if(rr.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION) {
                   if(rr.reply.id == AMQP_CHANNEL_CLOSE_METHOD) {
                           amqp_channel_close_t *m = (amqp_channel_close_t *)rr.reply.decoded;
                           if(m->reply_text.len > 0 && m->reply_text.bytes) {
                                   error_msg = (const char*)m->reply_text.bytes;
                                   error_len = m->reply_text.len;
                           }
                   } else if(rr.reply.id == AMQP_CONNECTION_CLOSE_METHOD) {
                           amqp_connection_close_t *m = (amqp_connection_close_t *)rr.reply.decoded;
                           if(m->reply_text.len > 0 && m->reply_text.bytes) {
                                   error_msg = (const char*)m->reply_text.bytes;
                                   error_len = m->reply_text.len;
                           }
                   }
           }
           
           if(error_msg) {
                   throw Exception("amqp", 0, "failed: %.*s", (int)error_len, error_msg);
           } else if(rr.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION) {
                   throw Exception("amqp", 0, "failed: library error %d", rr.library_error);
           } else {
                   throw Exception("amqp", 0, "failed");
           }
 }  }
   
   #define AMQP_STRING(s,l) new String(String::C(pa_strdup((const char*)(s), (l)), (l)))
   #define AMQP_VSTRING(s,l) new VString(*AMQP_STRING(s,l))
   
 static void _publish(Request& r, MethodParams& params) {  static void _publish(Request& r, MethodParams& params) {
         VAmqp& self=GET_SELF(r, VAmqp);          VAmqp& self=GET_SELF(r, VAmqp);
         const String &msg=params.as_string(0, "msg must be string");          const String &msg=params.as_string(0, "msg must be string");
         const char* exchange_c = ""; // default exchange          const char* exchange_c = ""; // default exchange
         const char* routing_key_c = 0;          const char* routing_key_c = 0;
         bool have_queue_sugar=false;  
         bool mandatory=false;          bool mandatory=false;
   
         amqp_basic_properties_t props;          amqp_basic_properties_t props;
Line 127  static void _publish(Request& r, MethodP Line 156  static void _publish(Request& r, MethodP
                                 } else if(key=="routing_key"){                                  } else if(key=="routing_key"){
                                         routing_key_c=value->as_string().cstr();                                          routing_key_c=value->as_string().cstr();
                                 } else if(key=="queue"){                                  } else if(key=="queue"){
                                         routing_key_c=value->as_string().cstr(); have_queue_sugar=true;                                          routing_key_c=value->as_string().cstr();
                                 } else if(key=="mandatory"){                                  } else if(key=="mandatory"){
                                         mandatory=r.process(*value).as_bool();                                          mandatory=r.process(*value).as_bool();
                                 } else if(key=="properties"){                                  } else if(key=="content_type"){
                                         // parse message properties                                          const char* v=value->as_string().cstr();
                                         if(HashStringValue* ph=value->get_hash()){                                          props.content_type=amqp_cstring_bytes(v);
                                                 for(HashStringValue::Iterator p(*ph); p; p.next()){                                          props._flags|=AMQP_BASIC_CONTENT_TYPE_FLAG;
                                                         String::Body pkey=p.key();                                  } else if(key=="content_encoding"){
                                                         Value* pval=p.value();                                          const char* v=value->as_string().cstr();
                                                         if(pkey=="content_type"){                                          props.content_encoding=amqp_cstring_bytes(v);
                                                                 const char* v=pval->as_string().cstr();                                          props._flags|=AMQP_BASIC_CONTENT_ENCODING_FLAG;
                                                                 props.content_type=amqp_cstring_bytes(v);                                  } else if(key=="delivery_mode"){
                                                                 props._flags|=AMQP_BASIC_CONTENT_TYPE_FLAG;                                          uint8_t dm=(uint8_t)value->as_int();
                                                         } else if(pkey=="content_encoding"){                                          props.delivery_mode=dm;
                                                                 const char* v=pval->as_string().cstr();                                          props._flags|=AMQP_BASIC_DELIVERY_MODE_FLAG;
                                                                 props.content_encoding=amqp_cstring_bytes(v);                                  } else if(key=="priority"){
                                                                 props._flags|=AMQP_BASIC_CONTENT_ENCODING_FLAG;                                          uint8_t pr=(uint8_t)value->as_int();
                                                         } else if(pkey=="delivery_mode"){                                          props.priority=pr;
                                                                 uint8_t dm=(uint8_t)pval->as_int();                                          props._flags|=AMQP_BASIC_PRIORITY_FLAG;
                                                                 props.delivery_mode=dm;                                  } else if(key=="correlation_id"){
                                                                 props._flags|=AMQP_BASIC_DELIVERY_MODE_FLAG;                                          const char* v=value->as_string().cstr();
                                                         } else if(pkey=="priority"){                                          props.correlation_id=amqp_cstring_bytes(v);
                                                                 uint8_t pr=(uint8_t)pval->as_int();                                          props._flags|=AMQP_BASIC_CORRELATION_ID_FLAG;
                                                                 props.priority=pr;                                  } else if(key=="reply_to"){
                                                                 props._flags|=AMQP_BASIC_PRIORITY_FLAG;                                          const char* v=value->as_string().cstr();
                                                         } else if(pkey=="correlation_id"){                                          props.reply_to=amqp_cstring_bytes(v);
                                                                 const char* v=pval->as_string().cstr();                                          props._flags|=AMQP_BASIC_REPLY_TO_FLAG;
                                                                 props.correlation_id=amqp_cstring_bytes(v);                                  } else if(key=="expiration"){
                                                                 props._flags|=AMQP_BASIC_CORRELATION_ID_FLAG;                                          const char* v=value->as_string().cstr();
                                                         } else if(pkey=="reply_to"){                                          props.expiration=amqp_cstring_bytes(v);
                                                                 const char* v=pval->as_string().cstr();                                          props._flags|=AMQP_BASIC_EXPIRATION_FLAG;
                                                                 props.reply_to=amqp_cstring_bytes(v);                                  } else if(key=="message_id"){
                                                                 props._flags|=AMQP_BASIC_REPLY_TO_FLAG;                                          const char* v=value->as_string().cstr();
                                                         } else if(pkey=="expiration"){                                          props.message_id=amqp_cstring_bytes(v);
                                                                 const char* v=pval->as_string().cstr();                                          props._flags|=AMQP_BASIC_MESSAGE_ID_FLAG;
                                                                 props.expiration=amqp_cstring_bytes(v);                                  } else if(key=="timestamp"){
                                                                 props._flags|=AMQP_BASIC_EXPIRATION_FLAG;                                          uint64_t ts=(uint64_t)value->as_double();
                                                         } else if(pkey=="message_id"){                                          props.timestamp=ts;
                                                                 const char* v=pval->as_string().cstr();                                          props._flags|=AMQP_BASIC_TIMESTAMP_FLAG;
                                                                 props.message_id=amqp_cstring_bytes(v);                                  } else if(key=="type"){
                                                                 props._flags|=AMQP_BASIC_MESSAGE_ID_FLAG;                                          const char* v=value->as_string().cstr();
                                                         } else if(pkey=="timestamp"){                                          props.type=amqp_cstring_bytes(v);
                                                                 uint64_t ts=(uint64_t)pval->as_double();                                          props._flags|=AMQP_BASIC_TYPE_FLAG;
                                                                 props.timestamp=ts;                                  } else if(key=="user_id"){
                                                                 props._flags|=AMQP_BASIC_TIMESTAMP_FLAG;                                          const char* v=value->as_string().cstr();
                                                         } else if(pkey=="type"){                                          props.user_id=amqp_cstring_bytes(v);
                                                                 const char* v=pval->as_string().cstr();                                          props._flags|=AMQP_BASIC_USER_ID_FLAG;
                                                                 props.type=amqp_cstring_bytes(v);                                  } else if(key=="app_id"){
                                                                 props._flags|=AMQP_BASIC_TYPE_FLAG;                                          const char* v=value->as_string().cstr();
                                                         } else if(pkey=="user_id"){                                          props.app_id=amqp_cstring_bytes(v);
                                                                 const char* v=pval->as_string().cstr();                                          props._flags|=AMQP_BASIC_APP_ID_FLAG;
                                                                 props.user_id=amqp_cstring_bytes(v);                                  } else if(key=="headers"){
                                                                 props._flags|=AMQP_BASIC_USER_ID_FLAG;  /*                                      if(HashStringValue* hh=pval->get_hash()){
                                                         } else if(pkey=="app_id"){                                                  size_t count=hh->count();
                                                                 const char* v=pval->as_string().cstr();                                                  amqp_table_entry_t* entries=count ? new amqp_table_entry_t[count] : 0;
                                                                 props.app_id=amqp_cstring_bytes(v);                                                  size_t idx=0;
                                                                 props._flags|=AMQP_BASIC_APP_ID_FLAG;                                                  for(HashStringValue::Iterator hi(*hh); hi; hi.next()){
                                                         } else if(pkey=="headers"){                                                          String::Body hkey=hi.key();
 /*                                                              if(HashStringValue* hh=pval->get_hash()){                                                          const char* hv=hi.value()->as_string().cstr();
                                                                         size_t count=hh->count();                                                          entries[idx].key=amqp_cstring_bytes(hkey.cstr());
                                                                         amqp_table_entry_t* entries=count ? new amqp_table_entry_t[count] : 0;                                                          entries[idx].value.kind=AMQP_FIELD_KIND_UTF8;
                                                                         size_t idx=0;                                                          entries[idx].value.value.bytes=amqp_cstring_bytes(hv);
                                                                         for(HashStringValue::Iterator hi(*hh); hi; hi.next()){                                                          idx++;
                                                                                 String::Body hkey=hi.key();  
                                                                                 const char* hv=hi.value()->as_string().cstr();  
                                                                                 entries[idx].key=amqp_cstring_bytes(hkey.cstr());  
                                                                                 entries[idx].value.kind=AMQP_FIELD_KIND_UTF8;  
                                                                                 entries[idx].value.value.bytes=amqp_cstring_bytes(hv);  
                                                                                 idx++;  
                                                                         }  
                                                                         props.headers.num_entries=(int)count;  
                                                                         props.headers.entries=entries;  
                                                                         props._flags|=AMQP_BASIC_HEADERS_FLAG;  
                                                                 }  
 */                                                      } else  
                                                                 throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);  
                                                 }                                                  }
                                                   props.headers.num_entries=(int)count;
                                                   props.headers.entries=entries;
                                                   props._flags|=AMQP_BASIC_HEADERS_FLAG;
                                         }                                          }
                                 } else  */                              } else
                                         throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);                                          throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
                         }                          }
                 }                  }
Line 218  static void _publish(Request& r, MethodP Line 237  static void _publish(Request& r, MethodP
         body.len = msg.length();          body.len = msg.length();
         body.bytes=(void*)msg.cstr();          body.bytes=(void*)msg.cstr();
   
         int ret = amqp_basic_publish(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), mandatory , 0, &props, body);          int ret = amqp_basic_publish(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), mandatory, 0, &props, body);
   
         if(ret!=AMQP_STATUS_OK)          if(ret!=AMQP_STATUS_OK)
                 throw Exception("amqp", 0, "publish failed");                  throw Exception("amqp", 0, "publish failed");
Line 227  static void _publish(Request& r, MethodP Line 246  static void _publish(Request& r, MethodP
         if(props._flags & AMQP_BASIC_HEADERS_FLAG){          if(props._flags & AMQP_BASIC_HEADERS_FLAG){
 //              delete [] props.headers.entries;  //              delete [] props.headers.entries;
         }          }
         (void)have_queue_sugar;  
 }  }
   
 static void _release(Request& r, MethodParams&) {  static void _release(Request& r, MethodParams&) {
Line 337  static void _declare_exchange(Request& r Line 355  static void _declare_exchange(Request& r
         if(!name_c)          if(!name_c)
                 throw Exception("amqp", 0, "name is required");                  throw Exception("amqp", 0, "name is required");
         amqp_exchange_declare(self.connection(), self.channel(), amqp_cstring_bytes(name_c), amqp_cstring_bytes(type_c), passive, durable, auto_delete, nowait, amqp_empty_table);          amqp_exchange_declare(self.connection(), self.channel(), amqp_cstring_bytes(name_c), amqp_cstring_bytes(type_c), passive, durable, auto_delete, nowait, amqp_empty_table);
         check("declare exchange", amqp_get_rpc_reply(self.connection()));          check(amqp_get_rpc_reply(self.connection()));
 }  }
   
 static void _delete_exchange(Request& r, MethodParams& params) {  static void _delete_exchange(Request& r, MethodParams& params) {
Line 362  static void _delete_exchange(Request& r, Line 380  static void _delete_exchange(Request& r,
         }          }
         if(!name_c) throw Exception("amqp", 0, "exchange is required");          if(!name_c) throw Exception("amqp", 0, "exchange is required");
         amqp_exchange_delete(self.connection(), self.channel(), amqp_cstring_bytes(name_c), if_unused);          amqp_exchange_delete(self.connection(), self.channel(), amqp_cstring_bytes(name_c), if_unused);
         check("delete exchange", amqp_get_rpc_reply(self.connection()));          check(amqp_get_rpc_reply(self.connection()));
 }  }
   
 static void _declare_queue(Request& r, MethodParams& params) {  static void _declare_queue(Request& r, MethodParams& params) {
Line 389  static void _declare_queue(Request& r, M Line 407  static void _declare_queue(Request& r, M
                 }                  }
         }          }
         amqp_queue_declare_ok_t *ok = amqp_queue_declare(self.connection(), self.channel(), queue_c ? amqp_cstring_bytes(queue_c) : amqp_empty_bytes, passive, durable, auto_delete, nowait, amqp_empty_table);          amqp_queue_declare_ok_t *ok = amqp_queue_declare(self.connection(), self.channel(), queue_c ? amqp_cstring_bytes(queue_c) : amqp_empty_bytes, passive, durable, auto_delete, nowait, amqp_empty_table);
         check("declare queue", amqp_get_rpc_reply(self.connection()));          check(amqp_get_rpc_reply(self.connection()));
         if(!queue_c && ok){          if(!queue_c && ok){
                 r.write(*new String(String::C(pa_strdup((const char *)ok->queue.bytes, ok->queue.len), ok->queue.len)));                  r.write(*AMQP_STRING(ok->queue.bytes, ok->queue.len));
         }          }
 }  }
   
Line 418  static void _delete_queue(Request& r, Me Line 436  static void _delete_queue(Request& r, Me
         }          }
         if(!queue_c) throw Exception("amqp", 0, "queue is required");          if(!queue_c) throw Exception("amqp", 0, "queue is required");
         amqp_queue_delete(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), if_unused, if_empty);          amqp_queue_delete(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), if_unused, if_empty);
         check("delete queue", amqp_get_rpc_reply(self.connection()));          check(amqp_get_rpc_reply(self.connection()));
 }  }
   
 static void _bind_queue(Request& r, MethodParams& params) {  static void _bind_queue(Request& r, MethodParams& params) {
Line 444  static void _bind_queue(Request& r, Meth Line 462  static void _bind_queue(Request& r, Meth
         }          }
         if(!exchange_c || !queue_c) throw Exception("amqp", 0, "exchange and queue are required");          if(!exchange_c || !queue_c) throw Exception("amqp", 0, "exchange and queue are required");
         amqp_queue_bind(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), amqp_empty_table);          amqp_queue_bind(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), amqp_empty_table);
         check("bind queue", amqp_get_rpc_reply(self.connection()));          check(amqp_get_rpc_reply(self.connection()));
 }  }
   
 static void _unbind_queue(Request& r, MethodParams& params) {  static void _unbind_queue(Request& r, MethodParams& params) {
Line 470  static void _unbind_queue(Request& r, Me Line 488  static void _unbind_queue(Request& r, Me
         }          }
         if(!exchange_c || !queue_c) throw Exception("amqp", 0, "exchange and queue are required");          if(!exchange_c || !queue_c) throw Exception("amqp", 0, "exchange and queue are required");
         amqp_queue_unbind(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), amqp_empty_table);          amqp_queue_unbind(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), amqp_empty_table);
         check("unbind queue", amqp_get_rpc_reply(self.connection()));          check(amqp_get_rpc_reply(self.connection()));
 }  }
   
 static void _consume(Request& r, MethodParams& params) {  static void _consume(Request& r, MethodParams& params) {
Line 480  static void _consume(Request& r, MethodP Line 498  static void _consume(Request& r, MethodP
         bool no_ack=true, nowait=false;          bool no_ack=true, nowait=false;
         Junction* callback=0;          Junction* callback=0;
   
         if(params.count()>0){          if(HashStringValue* options=params.as_hash(0)){
                 if(HashStringValue* options=params.as_hash(0)){                  for(HashStringValue::Iterator i(*options); i; i.next()){
                         for(HashStringValue::Iterator i(*options); i; i.next()){                          String::Body key=i.key();
                                 String::Body key=i.key();                          Value* value=i.value();
                                 Value* value=i.value();                          if(key=="callback"){
                                 if(key=="queue"){                                  callback=value->get_junction();
                                         queue_c=value->as_string().cstr();                          } else if(key=="queue"){
                                 } else if(key=="consumer_tag"){                                  queue_c=value->as_string().cstr();
                                         consumer_tag_c=value->as_string().cstr();                          } else if(key=="consumer_tag"){
                                 } else if(key=="no_ack"){                                  consumer_tag_c=value->as_string().cstr();
                                         no_ack=r.process(*value).as_bool();                          } else if(key=="no_ack"){
                                 } else if(key=="nowait"){                                  no_ack=r.process(*value).as_bool();
                                         nowait=r.process(*value).as_bool();                          } else if(key=="nowait"){
                                 } else if(key=="callback"){                                  nowait=r.process(*value).as_bool();
                                         callback=value->get_junction();                          } else
                                 } else                                  throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
                                         throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);  
                         }  
                 }                  }
         }          }
   
Line 507  static void _consume(Request& r, MethodP Line 523  static void _consume(Request& r, MethodP
         amqp_basic_consume(self.connection(), self.channel(), amqp_cstring_bytes(queue_c),          amqp_basic_consume(self.connection(), self.channel(), amqp_cstring_bytes(queue_c),
                 consumer_tag_c ? amqp_cstring_bytes(consumer_tag_c) : amqp_empty_bytes,                  consumer_tag_c ? amqp_cstring_bytes(consumer_tag_c) : amqp_empty_bytes,
                 0 /*no_local*/, no_ack, nowait, amqp_empty_table);                  0 /*no_local*/, no_ack, nowait, amqp_empty_table);
         amqp_rpc_reply_t rr = amqp_get_rpc_reply(self.connection());          check(amqp_get_rpc_reply(self.connection()));
         if(rr.reply_type != AMQP_RESPONSE_NORMAL)  
                 throw Exception("amqp", 0, "consume failed");  
   
         self.fstop=false;          self.fstop=false;
         while(!self.fstop){          while(!self.fstop){
                 amqp_envelope_t envelope; amqp_maybe_release_buffers(self.connection());                  amqp_envelope_t envelope;
                   memset(&envelope, 0, sizeof(envelope));
                   amqp_maybe_release_buffers(self.connection());
                 amqp_rpc_reply_t res = amqp_consume_message(self.connection(), &envelope, NULL, 0);                  amqp_rpc_reply_t res = amqp_consume_message(self.connection(), &envelope, NULL, 0);
                 if(res.reply_type == AMQP_RESPONSE_NORMAL){                  if(res.reply_type == AMQP_RESPONSE_NORMAL){
                         VHash &vh=*new VHash; HashStringValue* h=vh.get_hash();                          VHash &vh=*new VHash; HashStringValue* h=vh.get_hash();
                         String msg(String::Body((const unsigned char*)envelope.message.body.bytes, envelope.message.body.len), String::L_CLEAN);                          h->put("msg", AMQP_VSTRING(envelope.message.body.bytes, envelope.message.body.len));
                         h->put("msg", new VString(msg));  
                         h->put("delivery_tag", new VString(String::Body::uitoa((unsigned long long)envelope.delivery_tag)));                          h->put("delivery_tag", new VString(String::Body::uitoa((unsigned long long)envelope.delivery_tag)));
                         h->put("consumer_tag", new VString(String::Body((const unsigned char*)envelope.consumer_tag.bytes, envelope.consumer_tag.len)));                          h->put("consumer_tag", AMQP_VSTRING(envelope.consumer_tag.bytes, envelope.consumer_tag.len));
                         h->put("exchange", new VString(String::Body((const unsigned char*)envelope.exchange.bytes, envelope.exchange.len)));                          h->put("exchange", AMQP_VSTRING(envelope.exchange.bytes, envelope.exchange.len));
   
                         Value *params_cb[]={&vh};                          Value *params_cb[]={&vh};
                         METHOD_FRAME_ACTION(*callback->method, r.method_frame, callback->self, {                          METHOD_FRAME_ACTION(*callback->method, r.method_frame, callback->self, {
Line 561  MAmqp::MAmqp(): Methoded("amqp") { Line 576  MAmqp::MAmqp(): Methoded("amqp") {
         add_native_method("delete_queue", Method::CT_DYNAMIC, _delete_queue, 0, 1);          add_native_method("delete_queue", Method::CT_DYNAMIC, _delete_queue, 0, 1);
         add_native_method("bind_queue", Method::CT_DYNAMIC, _bind_queue, 0, 1);          add_native_method("bind_queue", Method::CT_DYNAMIC, _bind_queue, 0, 1);
         add_native_method("unbind_queue", Method::CT_DYNAMIC, _unbind_queue, 0, 1);          add_native_method("unbind_queue", Method::CT_DYNAMIC, _unbind_queue, 0, 1);
         add_native_method("consume", Method::CT_DYNAMIC, _consume, 0, 1);          add_native_method("consume", Method::CT_DYNAMIC, _consume, 1, 1);
         add_native_method("stop_consume", Method::CT_DYNAMIC, _stop_consume, 0, 0);          add_native_method("stop_consume", Method::CT_DYNAMIC, _stop_consume, 0, 0);
 #endif  #endif
 }  }
   
   
Removed from v.1.1  
changed lines
  Added in v.1.4

E-mail: