amqpnodesexchange.cpp

#include "amqpnodesexchange.h"

AmqpNodesExchange::AmqpNodesExchange(QObject *parent) : QObject(parent)
{
    connectToMq();
}

void AmqpNodesExchange::connectToMq()
{
    connect(&m_client, SIGNAL(connected()), this, SLOT(on_clientConnected()));
    connect(this, SIGNAL(exchangeReady()), this, SLOT(on_exchangeReady()));

    QSettings s;
    m_client.setUsername(s.value("mq_user").toString());
    m_client.setPassword(s.value("mq_pw").toString());
    m_client.connectToHost(QHostAddress(s.value("mq_host").toString()));
    //m_client.setAutoReconnect(true);
}

void AmqpNodesExchange::on_clientConnected()
{
    createExchange();
}

void AmqpNodesExchange::createExchange()
{
    m_nodesExchange = m_client.createExchange("direct_nodes");
    m_nodesExchange->declare(QAmqpExchange::Direct, QAmqpExchange::Durable);
    m_nodesExchange->enableConfirms();
    connect(m_nodesExchange, SIGNAL(declared()), this, SLOT(on_exchangeReady()));
}

void AmqpNodesExchange::on_exchangeReady()
{
    qDebug() << "Nodes Exchange declared";
}

void AmqpNodesExchange::writeMessage(QString nodeId, QByteArray msg)
{
    QAmqpMessage::PropertyHash properties;
    properties[QAmqpMessage::ContentType] = "application/json";

    m_nodesExchange->publish(msg, "node_"+nodeId, properties);
    //m_nodesExchange->waitForConfirms();
}