ZMQ Python PUB / SUB работает, но не мой подписчик C ++ с Python Publisher - PullRequest
1 голос
/ 05 апреля 2019

Я использую Ubuntu 18.04, Qt 5.12 и libzmq.so.5.1.5.У меня есть 2 очень простых скрипта Python 3.6, использующих pyzmq 18.0.1, один реализует ZMQ PUB и один реализует ZMQ SUB.Эти скрипты Python работают, находят, и данные передаются от одного к другому в локальной сети обратной связи.Я пытаюсь внедрить подписчика в Qt 5.12, и подписчик всегда блокирует прием, когда я отправляю одни и те же данные, используя один и тот же скрипт издателя Python.Я вставил полный код ниже - как я могу получить ZMQ с Qt 5.12 для фактически полученных данных?Спасибо.

Вывод моего приложения на C ++ выглядит успешным:

"Server IP Address determined to be: 127.0.0.1"
"Connecting to: tcp://127.0.0.1:5555"
"Attempted Connect: 0"
"SetSockOpt: 0"
"GetSockOpt: 0"
Done setting up socket
Waiting

Python 3 PUB:

#!/usr/bin/python3

import time
import zmq
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://127.0.0.1:5555")

# Allow clients to connect before sending data
time.sleep(3)
while True:
    socket.send_pyobj({1:[1,2,3]})
    time.sleep(1)

Python 3 SUB:

#!/usr/bin/python3

import zmq
context = zmq.Context()
socket = context.socket(zmq.SUB)
# We can connect to several endpoints if we desire, and receive from all.
socket.connect("tcp://127.0.0.1:5555")

# We must declare the socket as of type SUBSCRIBER, and pass a prefix filter.
# Here, the filter is the empty string, wich means we receive all messages.
# We may subscribe to several filters, thus receiving from all.
socket.setsockopt(zmq.SUBSCRIBE, b'')

while True:
    message = socket.recv_pyobj()
    print(message.get(1)[2])

Qt 5.12 Код абонента:

#include <QCoreApplication>

// Std includes
#include <stdio.h>
#include <unistd.h>
#include <assert.h>

// Qt
#include <QDebug>
#include <QFile>

// ZeroMQ Includes
#include <zmq.h>

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    //QString m_ServerIP = QString("*");
    QString m_ServerIP = QString("127.0.0.1");
    QString m_ServerPort = QString("5555");

    qDebug() << QString("Server IP Address determined to be: %1").arg(m_ServerIP);

    void* m_Context = zmq_ctx_new();
    assert (m_Context);
    void* m_Subscriber = zmq_socket (m_Context, ZMQ_SUB);
    assert (m_Subscriber);
    int rc = -1;
    unsigned int fd = 0;
    do {

        const char *filter = std::string("").c_str();
        QString ipAndPort = QString("tcp://%1:%2").arg(m_ServerIP).arg(m_ServerPort);

        qDebug() << QString("Connecting to: %1").arg(ipAndPort);

        rc = zmq_connect(m_Subscriber, ipAndPort.toStdString().c_str());

        qDebug() << QString("Attempted Connect: %1").arg(rc);

        rc = zmq_setsockopt(m_Subscriber, ZMQ_SUBSCRIBE,filter, strlen (filter));

        qDebug() << QString("SetSockOpt: %1").arg(rc);

        size_t fd_size = sizeof(fd);

        rc = zmq_getsockopt(m_Subscriber,ZMQ_FD,&fd,&fd_size);

        qDebug() << QString("GetSockOpt: %1").arg(rc);

    }
    while ( rc < 0 );

    qDebug() << "Done setting up socket";

    while ( true) {

        zmq_msg_t message;
        zmq_msg_init(&message);
        qDebug() << "Waiting";
        zmq_recvmsg(m_Subscriber, &message, 0);
        size_t size = zmq_msg_size (&message);

        qDebug() << QString("Message Size: %1").arg(size);

        char *string = static_cast<char*>(malloc(size + 1));
        memcpy (string, zmq_msg_data(&message), size);
        zmq_msg_close (&message);
        string [size] = 0;
        if (string) {
            QByteArray frame = QByteArray::fromBase64(QByteArray(string));
            free(string);

            qDebug() << QString("Debug RX Frame Size: %1").arg(frame.size());

            QFile output("/tmp/abcd.jpeg");
            if ( output.open(QIODevice::WriteOnly) ) {
                output.write(frame);
                output.close();
            }

        }

    }

    return a.exec();
}

1 Ответ

1 голос
/ 06 апреля 2019

У вас есть следующие ошибки:

  • Вам не нужно QCoreApplication, поскольку вы не используете сигналы, слоты, события и т. Д., Для которых требуется цикл обработки событий.

  • Вы используете C api zmq, есть совместимый API для C ++, и для его использования используйте заголовок <zmq.hpp>.

  • Когда выиспользуйте send_pyobj() python, он использует pickle для передачи данных, но в C ++ нет способа вернуть данные.Вместо этого вы должны отправить данные с send_json() из python.

Учитывая вышеизложенное, решение:

*. Py

#!/usr/bin/python3

import time
import zmq
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://127.0.0.1:5555")

# Allow clients to connect before sending data
time.sleep(3)
while True:
    socket.send_json({1:[1,2,3]})
    time.sleep(1)

main.cpp

#include <QString>
#include <QDebug>
#include <QUrl>
#include <QThread>

#include <zmq.hpp>

int main(void) {

    QString m_ServerIP = QString("127.0.0.1");
    int m_ServerPort = 5555;

    qDebug() << QString("Server IP Address determined to be: %1").arg(m_ServerIP);

    QUrl url;
    url.setScheme("tcp");
    url.setHost(m_ServerIP);
    url.setPort(m_ServerPort);

    zmq::context_t context(1);
    zmq::socket_t subscriber (context, ZMQ_SUB);
    subscriber.connect(url.toString().toStdString());
    subscriber.setsockopt( ZMQ_SUBSCRIBE, "", 0);

    while(true) {
        zmq_msg_t in_msg;
        zmq_msg_init(&in_msg);
        zmq::message_t  message;
        subscriber.recv(&message);
        qDebug() << QString("Message Size: %1").arg(message.size());
        QByteArray ba(static_cast<char*>(message.data()), message.size());
        qDebug()<< "message" << ba;
        QThread::sleep(1);
    }
}

Вывод:

"Server IP Address determined to be: 127.0.0.1"
"Message Size: 13"
message "{\"1\":[1,2,3]}"
"Message Size: 13"
message "{\"1\":[1,2,3]}"
"Message Size: 13"
message "{\"1\":[1,2,3]}"
"Message Size: 13"
message "{\"1\":[1,2,3]}"
"Message Size: 13"
message "{\"1\":[1,2,3]}"

Вы можете декодировать данные, используя QJsonDocument .Я также рекомендую вам просмотреть следующие примеры на C ++ из zmq .


Если вы используете socket.send_pyobj({1:[1,2,3]}), вы получите следующее, что является результатом pickle:

"Server IP Address determined to be: 127.0.0.1"
"Message Size: 20"
message "\x80\x03}q\x00K\x01]q\x01(K\x01K\x02K\x03""es."
"Message Size: 20"
message "\x80\x03}q\x00K\x01]q\x01(K\x01K\x02K\x03""es."
"Message Size: 20"
message "\x80\x03}q\x00K\x01]q\x01(K\x01K\x02K\x03""es."
"Message Size: 20"
message "\x80\x03}q\x00K\x01]q\x01(K\x01K\x02K\x03""es."
"Message Size: 20"
message "\x80\x03}q\x00K\x01]q\x01(K\x01K\x02K\x03""es."
...