jessequ
/
AMQP-CPP
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

removed code that is useless for others, fixes #248)

This commit is contained in:
Emiel Bruijntjes 2018-08-29 19:33:06 +02:00
parent 3d33332c23
commit 9661f290a6
4 changed files with 0 additions and 402 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tests/README.md
View File

@ -1,2 +0,0 @@
I'm sorry, the test case makes use of the closed source Copernica libraries. Maybe someone
is willing to provide a test case based on plain system calls?

54
tests/main.cpp
View File

@ -1,54 +0,0 @@
/**
* Main.cpp
*
* Test program
*
* @copyright 2014 Copernica BV
*/
/**
* Global libraries that we need
*/
#include <amqpcpp.h>
#include <copernica/network.h>
/**
* Namespaces to use
*/
using namespace std;
using namespace Copernica;
/**
* Local libraries
*/
#include "myconnection.h"
/**
* Main procedure
* @param argc
* @param argv
* @return int
*/
int main(int argc, const char *argv[])
{
// need an ip
if (argc != 2)
{
// report error
std::cerr << "usage: " << argv[0] << " <ip>" << std::endl;
// done
return -1;
}
else
{
// create connection
MyConnection connection(argv[1]);
// start the main event loop
Event::MainLoop::instance()->run();
// done
return 0;
}
}

228
tests/myconnection.cpp
View File

@ -1,228 +0,0 @@
/**
* MyConnection.cpp
*
* @copyright 2014 Copernica BV
*/
/**
* Required external libraries
*/
#include <amqpcpp.h>
#include <copernica/network.h>
#include <string>
/**
* Namespaces to use
*/
using namespace std;
using namespace Copernica;
/**
* Required local class definitions
*/
#include "myconnection.h"
/**
* Constructor
*/
MyConnection::MyConnection(const std::string &ip) :
_socket(Event::MainLoop::instance(), this)
{
// start connecting
if (_socket.connect(Dns::IpAddress(ip), 5672)) return;
// failure
onFailure(&_socket);
}
/**
* Destructor
*/
MyConnection::~MyConnection()
{
}
/**
* Method that is called when the connection failed
* @param socket Pointer to the socket
*/
void MyConnection::onFailure(Network::TcpSocket *socket)
{
// report error
std::cout << "connect failure" << std::endl;
}
/**
* Method that is called when the connection timed out (which also is a failure
* @param socket Pointer to the socket
*/
void MyConnection::onTimeout(Network::TcpSocket *socket)
{
// report error
std::cout << "connect timeout" << std::endl;
}
/**
* Method that is called when the connection succeeded
* @param socket Pointer to the socket
*/
void MyConnection::onConnected(Network::TcpSocket *socket)
{
// report connection
std::cout << "connected" << std::endl;
// we are connected, leap out if there already is a amqp connection
if (_connection) return;
// create amqp connection, and a new channel
_connection = std::make_shared<AMQP::Connection>(this, AMQP::Login("guest", "guest"), "/");
_channel = std::make_shared<AMQP::Channel>(_connection.get());
// install a handler when channel is in error
_channel->onError([](const char *message) {
std::cout << "channel error " << message << std::endl;
});
// install a handler when channel is ready
_channel->onReady([]() {
std::cout << "channel ready" << std::endl;
});
// we declare a queue, an exchange and we publish a message
_channel->declareQueue("my_queue").onSuccess([this]() {
std::cout << "queue declared" << std::endl;
// start consuming
_channel->consume("my_queue").onReceived([this](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered) {
std::cout << "consumed from exchange " << message.exchange() << " " << message.routingkey() << ": " << std::string(message.body(), message.bodySize()) << std::endl;
_channel->ack(deliveryTag);
});
});
// declare an exchange
_channel->declareExchange("my_exchange", AMQP::direct).onSuccess([]() {
std::cout << "exchange declared" << std::endl;
});
// bind queue and exchange
_channel->bindQueue("my_exchange", "my_queue", "key").onSuccess([this]() {
std::cout << "queue bound to exchange" << std::endl;
// callback for returns
auto callback = [](const AMQP::Message &message, int16_t code, const std::string &description) {
std::cout << "message was returned: " << code << " " << description << ": " << std::string(message.body(), message.bodySize()) << std::endl;
};
_channel->publish("my_exchange", "key", "just a message", AMQP::mandatory).onReturned(callback);
_channel->publish("my_exchange", "unknown key", "just another message", AMQP::mandatory).onReturned(callback);
});
}
/**
* Method that is called when the socket is closed (as a result of a TcpSocket::close() call)
* @param socket Pointer to the socket
*/
void MyConnection::onClosed(Network::TcpSocket *socket)
{
// show
std::cout << "myconnection closed" << std::endl;
// set to null
_channel = nullptr;
_connection = nullptr;
}
/**
* Method that is called when the peer closed the connection
* @param socket Pointer to the socket
*/
void MyConnection::onLost(Network::TcpSocket *socket)
{
// report error
std::cout << "connection lost" << std::endl;
// set to null
_channel = nullptr;
_connection = nullptr;
}
/**
* Method that is called when data is received on the socket
* @param socket Pointer to the socket
* @param buffer Pointer to the fill input buffer
*/
void MyConnection::onData(Network::TcpSocket *socket, Network::Buffer *buffer)
{
// send what came in
std::cout << "received: " << buffer->size() << " bytes" << std::endl;
// leap out if there is no connection
if (!_connection) return;
// let the data be handled by the connection
size_t bytes = _connection->parse(buffer->buffer(), buffer->size());
// shrink the buffer
buffer->shrink(bytes);
}
/**
* Method that is called when data needs to be sent over the network
*
* Note that the AMQP library does no buffering by itself. This means
* that this method should always send out all data or do the buffering
* itself.
*
* @param connection The connection that created this output
* @param buffer Data to send
* @param size Size of the buffer
*/
void MyConnection::onData(AMQP::Connection *connection, const char *buffer, size_t size)
{
// // report what is going on
// std::cout << "send: " << size << std::endl;
//
// for (unsigned i=0; i<size; i++) std::cout << (int)buffer[i] << " ";
// std::cout << std::endl;
// send to the socket
_socket.write(buffer, size);
}
/**
* When the connection ends up in an error state this method is called.
* This happens when data comes in that does not match the AMQP protocol
*
* After this method is called, the connection no longer is in a valid
* state and can be used. In normal circumstances this method is not called.
*
* @param connection The connection that entered the error state
* @param message Error message
*/
void MyConnection::onError(AMQP::Connection *connection, const char *message)
{
// report error
std::cout << "AMQP Connection error: " << message << std::endl;
}
/**
* Method that is called when the login attempt succeeded. After this method
* was called, the connection is ready to use
*
* @param connection The connection that can now be used
*/
void MyConnection::onConnected(AMQP::Connection *connection)
{
// show
std::cout << "AMQP login success" << std::endl;
// create channel if it does not yet exist
if (!_channel) _channel = std::make_shared<AMQP::Channel>(connection);
}

118
tests/myconnection.h
View File

@ -1,118 +0,0 @@
/**
* MyConnection.h
*
* Our own test implementation for a connection handler
*
* @copyright 2014 Copernica BV
*/
/**
* Class definition
*/
class MyConnection :
public AMQP::ConnectionHandler,
public Network::TcpHandler
{
private:
/**
* The actual TCP socket that is connected with RabbitMQ
* @var TcpSocket
*/
Network::TcpSocket _socket;
/**
* The AMQP connection
* @var Connection
*/
std::shared_ptr<AMQP::Connection> _connection;
/**
* The AMQP channel
* @var Channel
*/
std::shared_ptr<AMQP::Channel> _channel;
/**
* Method that is called when the connection failed
* @param socket Pointer to the socket
*/
virtual void onFailure(Network::TcpSocket *socket) override;
/**
* Method that is called when the connection timed out (which also is a failure
* @param socket Pointer to the socket
*/
virtual void onTimeout(Network::TcpSocket *socket) override;
/**
* Method that is called when the connection succeeded
* @param socket Pointer to the socket
*/
virtual void onConnected(Network::TcpSocket *socket) override;
/**
* Method that is called when the socket is closed (as a result of a TcpSocket::close() call)
* @param socket Pointer to the socket
*/
virtual void onClosed(Network::TcpSocket *socket) override;
/**
* Method that is called when the peer closed the connection
* @param socket Pointer to the socket
*/
virtual void onLost(Network::TcpSocket *socket) override;
/**
* Method that is called when data is received on the socket
* @param socket Pointer to the socket
* @param buffer Pointer to the fill input buffer
*/
virtual void onData(Network::TcpSocket *socket, Network::Buffer *buffer) override;
/**
* Method that is called when data needs to be sent over the network
*
* Note that the AMQP library does no buffering by itself. This means
* that this method should always send out all data or do the buffering
* itself.
*
* @param connection The connection that created this output
* @param buffer Data to send
* @param size Size of the buffer
*/
virtual void onData(AMQP::Connection *connection, const char *buffer, size_t size) override;
/**
* When the connection ends up in an error state this method is called.
* This happens when data comes in that does not match the AMQP protocol
*
* After this method is called, the connection no longer is in a valid
* state and can be used. In normal circumstances this method is not called.
*
* @param connection The connection that entered the error state
* @param message Error message
*/
virtual void onError(AMQP::Connection *connection, const char *message) override;
/**
* Method that is called when the login attempt succeeded. After this method
* was called, the connection is ready to use
*
* @param connection The connection that can now be used
*/
virtual void onConnected(AMQP::Connection *connection) override;
public:
/**
* Constructor
* @param ip
*/
MyConnection(const std::string &ip);
/**
* Destructor
*/
virtual ~MyConnection();
};