Как использовать ZeroMQ в приложении GTK / QT / Clutter? - PullRequest
Новая
       3

Как использовать ZeroMQ в приложении GTK / QT / Clutter?

5 голосов
/ 23 июня 2011

В gtk приложениях все выполнение происходит внутри функции gtk_main.И другие графические фреймы имеют похожие циклы событий, такие как app.exec для QT и clutter_main для Clutter .Однако ZeroMQ основан на предположении, что есть цикл while (1) ..., в который он вставлен (см., Например, здесь для примеров).

Как выобъединить эти две стратегии выполнения?

В настоящее время я хочу использовать zeromq в беспорядочном приложении, написанном на C, поэтому я, конечно, хотел бы получить прямые ответы на этот вопрос, но, пожалуйста, добавьте ответы и для других вариантов.

Ответы [ 5 ]

12 голосов
/ 02 июня 2012

Правильный способ объединить zmq и gtk или беспорядок - это подключить дескриптор файла очереди zmq к главному циклу событий. FD можно получить с помощью 

int fd;
size_t sizeof_fd = sizeof(fd);
if(zmq_getsockopt(socket, ZMQ_FD, &fd, &sizeof_fd))
      perror("retrieving zmq fd");

Подключение его к основному циклу - это вопрос использования io_add_watch: 

GIOChannel* channel = g_io_channel_unix_new(fd);    
g_io_add_watch(channel, G_IO_IN|G_IO_ERR|G_IO_HUP, callback_func, NULL);

В функции обратного вызова перед проверкой необходимо сначала проверить, действительно ли есть что читать. В противном случае функция может заблокировать ожидание ввода-вывода. 

gboolean callback_func(GIOChannel *source, GIOCondition condition,gpointer data)
{
    uint32_t status;
    size_t sizeof_status = sizeof(status);   

    while (1){
         if (zmq_getsockopt(socket, ZMQ_EVENTS, &status, &sizeof_status)) {
             perror("retrieving event status");
             return 0; // this just removes the callback, but probably
                       // different error handling should be implemented
         }
         if (status & ZMQ_POLLIN == 0) {
             break;
         }

         // retrieve one message here
    }
    return 1; // keep the callback active
}

Пожалуйста, обратите внимание: это на самом деле не проверено, я сделал перевод с Python + Clutter, который я и использую, но я уверен, что он будет работать. Для справки ниже приведен полный код Python + Clutter, который на самом деле работает. 

import sys
from gi.repository import Clutter, GObject
import zmq

def Stage():
    "A Stage with a red spinning rectangle"
    stage = Clutter.Stage()

    stage.set_size(400, 400)
    rect = Clutter.Rectangle()
    color = Clutter.Color()
    color.from_string('red')
    rect.set_color(color)
    rect.set_size(100, 100)
    rect.set_position(150, 150)

    timeline = Clutter.Timeline.new(3000)
    timeline.set_loop(True)

    alpha = Clutter.Alpha.new_full(timeline, Clutter.AnimationMode.EASE_IN_OUT_SINE)
    rotate_behaviour = Clutter.BehaviourRotate.new(
        alpha, 
        Clutter.RotateAxis.Z_AXIS,
        Clutter.RotateDirection.CW,
        0.0, 359.0)
    rotate_behaviour.apply(rect)
    timeline.start()
    stage.add_actor(rect)

    stage.show_all()
    stage.connect('destroy', lambda stage: Clutter.main_quit())
    return stage, rotate_behaviour

def Socket(address):
    ctx = zmq.Context()
    sock = ctx.socket(zmq.SUB)
    sock.setsockopt(zmq.SUBSCRIBE, "")
    sock.connect(address)
    return sock

def zmq_callback(queue, condition, sock):
    print 'zmq_callback', queue, condition, sock

    while sock.getsockopt(zmq.EVENTS) & zmq.POLLIN:
        observed = sock.recv()
        print observed

    return True

def main():
    res, args = Clutter.init(sys.argv)
    if res != Clutter.InitError.SUCCESS:
        return 1

    stage, rotate_behaviour = Stage()

    sock = Socket(sys.argv[2])
    zmq_fd = sock.getsockopt(zmq.FD)
    GObject.io_add_watch(zmq_fd,
                         GObject.IO_IN|GObject.IO_ERR|GObject.IO_HUP,
                         zmq_callback, sock)

    return Clutter.main()

if __name__ == '__main__':
    sys.exit(main())
6 голосов
/ 23 июня 2011

Звучит так, будто код ZeroMQ хочет просто выполнять снова и снова как можно чаще.Самый простой способ - поместить код ZeroMQ в функцию ожидания или функцию тайм-аута и использовать неблокирующие версии функций, если они существуют.

Для Clutter вы должны использовать clutter_threads_add_idle() или clutter_threads_add_timeout().Для GTK вы должны использовать g_idle_add() или g_timeout_add().

Более сложный, но, возможно, лучший способ - создать отдельный поток для ZeroMQ.код, используя g_thread_create(), и просто используйте конструкцию while(1) с функциями блокировки, как они предлагают.Если вы сделаете это, вам также придется найти способ взаимодействия потоков друг с другом - мьютексы GLib и асинхронные очереди обычно работают нормально.

2 голосов
/ 27 июля 2011

Вы можете получить дескриптор файла для сокета 0MQ (опция ZMQ_FD) и интегрировать его с вашим циклом событий. Я предполагаю, что у gtk есть некоторый механизм для работы с сокетами.

2 голосов
/ 23 июня 2011

Я обнаружил, что есть библиотека интеграции QT с именем Zeromqt . Глядя на источник, ядро ​​интеграции заключается в следующем: 

ZmqSocket::ZmqSocket(int type, QObject *parent) : QObject(parent)
{
    ...
    notifier_ = new QSocketNotifier(fd, QSocketNotifier::Read, this);
    connect(notifier_, SIGNAL(activated(int)), this, SLOT(activity()));
}

...

void ZmqSocket::activity()
{
    uint32_t flags;
    size_t size = sizeof(flags);
    if(!getOpt(ZMQ_EVENTS, &flags, &size)) {
        qWarning("Error reading ZMQ_EVENTS in ZMQSocket::activity");
        return;
    }
    if(flags & ZMQ_POLLIN) {
        emit readyRead();
    }
    if(flags & ZMQ_POLLOUT) {
        emit readyWrite();
    }
    ...
}

Следовательно, он полагается на интегрированную обработку сокетов QT, и Clutter не будет иметь ничего подобного.

1 голос
/ 17 августа 2015

Это пример в Python, использующий PyQt4. Он получен из рабочего приложения. 

import zmq
from PyQt4 import QtCore, QtGui

class QZmqSocketNotifier( QtCore.QSocketNotifier ):
    """ Provides Qt event notifier for ZMQ socket events """
    def __init__( self, zmq_sock, event_type, parent=None ):
        """
        Parameters:
        ----------
        zmq_sock : zmq.Socket
            The ZMQ socket to listen on. Must already be connected or bound to a socket address.
        event_type : QtSocketNotifier.Type
            Event type to listen for, as described in documentation for QtSocketNotifier.
        """
        super( QZmqSocketNotifier, self ).__init__( zmq_sock.getsockopt(zmq.FD), event_type, parent )

class Server(QtGui.QFrame):

def __init__(self, topics, port, mainwindow, parent=None):
    super(Server, self).__init__(parent)

    self._PORT = port

    # Create notifier to handle ZMQ socket events coming from client
    self._zmq_context = zmq.Context()
    self._zmq_sock = self._zmq_context.socket( zmq.SUB )
    self._zmq_sock.bind( "tcp://*:" + self._PORT )
    for topic in topics:
        self._zmq_sock.setsockopt( zmq.SUBSCRIBE, topic )
    self._zmq_notifier = QZmqSocketNotifier( self._zmq_sock, QtCore.QSocketNotifier.Read )

    # connect signals and slots
    self._zmq_notifier.activated.connect( self._onZmqMsgRecv )
    mainwindow.quit.connect( self._onQuit )

@QtCore.pyqtSlot()
def _onZmqMsgRecv():
    self._test_info_notifier.setEnabled(False)
    # Verify that there's data in the stream
    sock_status = self._zmq_sock.getsockopt( zmq.EVENTS )
    if sock_status == zmq.POLLIN:
        msg = self._zmq_sock.recv_multipart()
        topic = msg[0]
        callback = self._topic_map[ topic ]
        callback( msg )
    self._zmq_notifier.setEnabled(True)
    self._zmq_sock.getsockopt(zmq.EVENTS)

def _onQuit(self):
    self._zmq_notifier.activated.disconnect( self._onZmqMsgRecv )
    self._zmq_notifier.setEnabled(False)
    del self._zmq_notifier
    self._zmq_context.destroy(0)

Отключение, а затем повторное включение уведомителя в _on_ZmqMsgRecv соответствует документации для QSocketNotifier .

Последний вызов по getockopt по какой-то причине необходим. В противном случае уведомитель перестает работать после первого события. Я собирался опубликовать новый вопрос для этого. Кто-нибудь знает, зачем это нужно?

Обратите внимание, что если вы не уничтожите уведомитель перед контекстом ZMQ, вы, вероятно, получите ошибку, подобную этой, при выходе из приложения: 

QSocketNotifier: Invalid socket 16 and type 'Read', disabling...
...