QAMQP：用Qt实现AMQP 0.9.1

jopen 10年前

AMQP 0.9.1的Qt4/Qt5实现

Implement

Connection

work with socket connections

  • start - start connection negotiation
  • startok - select security mechanism and locale
  • tune - propose connection tuning parameters
  • tuneok - negotiate connection tuning parameters
  • open - open connection to virtual host
  • openok - signal that connection is ready
  • close - request a connection close
  • closeok - confirm a connection close

Channel

work with channels

  • open - open a channel for use
  • openok - signal that the channel is ready
  • close - request a channel close
  • closeok - confirm a channel close

Exchange

work with exchanges

  • declare - verify exchange exists, create if needed
  • declareok - confirm exchange declaration
  • delete - delete an exchange
  • deleteok - confirm deletion of an exchange

Queue

work with queues

  • declare - declare queue, create if needed
  • declareok - confirms a queue definition
  • bind - bind queue to an exchange
  • bindok - confirm bind successful
  • unbind - unbind a queue from an exchange
  • unbindok - confirm unbind successful
  • purge - purge a queue
  • purgeok - confirms a queue purge
  • delete - delete a queue
  • deleteok - confirm deletion of a queue

Basic

work with basic content

  • qos - specify quality of service
  • qosok - confirm the requested qos
  • consume - start a queue consumer
  • consumeok - confirm a new consumer
  • publish - publish a message
  • deliver - notify the client of a consumer message
  • get - direct access to a queue
  • getok - provide client with a message
  • getempty - indicate no messages available
  • ack - acknowledge one or more messages

Usage
Test::Test()      {      QUrl con(QString("amqp://guest:guest@localhost:5672/"));      client_ = new QAMQP::Client(this);      connect(client_, SIGNAL(connected()), this, SLOT(connected()));      client_->open(con);      exchange_ =  client_->createExchange("test.test2");      queue_ = client_->createQueue("test.my_queue", exchange_->channelNumber());        connect(queue_, SIGNAL(declared()), this, SLOT(declared()));      connect(queue_, SIGNAL(messageReceived(QAMQP::Queue * )), this, SLOT(newMessage(QAMQP::Queue *)));      }    void Test::connected()  {      exchange_->declare("fanout");             queue_->declare();      exchange_->bind(queue_);  }    void Test::declared()  {      exchange_->publish("Hello world", exchange_->name());      queue_->setQOS(0,10);      queue_->setConsumerTag("qamqp-consumer");      queue_->consume(QAMQP::Queue::coNoAck);  }    void Test::newMessage(QAMQP::Queue * q)  {      while (q->hasMessage())      {          QAMQP::MessagePtr message = q->getMessage();          qDebug("+ RECEIVE MESSAGE");          qDebug("| Exchange-name: %s", qPrintable(message->exchangeName));          qDebug("| Routing-key: %s", qPrintable(message->routeKey));          qDebug("| Content-type: %s", qPrintable(message->property[QAMQP::Frame::Content::cpContentType].toString()));          if(!q->noAck())          {              q->ack(message);          }      }  }

项目主页：http://www.open-open.com/lib/view/home/1404197023811