Как выполнить архетип ZeroMQ PUSH / PULL асинхронным способом? - PullRequest
Новая
       51

Как выполнить архетип ZeroMQ PUSH / PULL асинхронным способом?

0 голосов
/ 17 апреля 2020

Я хочу инициировать PULL в порту и хочу получать данные с других портов на мой порт PULL. В случае порта PULL он слушает асинхронно, а когда получает сообщение, он просто печатает сообщение в консоли. Поэтому для этого я написал метод внутри Push -класса, который отправит сообщение на порт PULL.

Мой код выглядит следующим образом: 

import random
import zmq
import time
import sys
import string
import asyncio
import zmq.asyncio

class Push():
    def __init__(self, port, addr='localhost'):
        self.port = port
        self.addr = addr

        self.ctx = zmq.Context()
        self.scoket = self.ctx.socket(zmq.PUSH)
        self.scoket.connect(f'tcp://{self.addr}:{selfa.port}')

    def send(self):
        chars = string.ascii_uppercase + string.ascii_lowercase
        message = ''.join(random.choice(chars) for _ in range(4))
        self.scoket.send(bytes(message, 'utf-8'))
        print(f'sending: {message}')

class Pull():
    def __init__(self, port, addr='*'):
        self.port = port
        self.addr = addr

        self.ctx = zmq.Context()
        self.socket = self.ctx.socket(zmq.PULL)
        self.socket.bind(f'tcp://{self.addr}:{self.port}')

    async def listen(self, listener):
        while True:
            string = await self.socket.recv()
            listener(string)


if __name__ == '__main__':
    push = Push('55501')

    async def send():
        while True:
            await asyncio.sleep(5)
            print('Sending...')
            push.send()

    pull = Pull('55501')

    try:
        asyncio.run(
            pull.listen(print),
            send(),
        )
    except KeyboardInterrupt:
        print('exiting...')
        exit()

вышеуказанный код не работает. Код останавливается на методе listen.

Ответы [ 2 ]

0 голосов
/ 18 апреля 2020

Приведенный выше код не работает.

Код выполняется,
, но в исходном коде есть ошибки (в концепции) и опечатки .

Код "как есть" запрещает классу Push стать .connect() -едом, таким образом, Pull -сопоставленная сторона .bind() -s, пока не с кем поговорить.

РЕШЕНИЕ

1)
Исправить эту опечатку (+ лучше, явно обнаруживать и обрабатывать все потенциальные состояния ошибки) 

self.scoket.connect(f'tcp://{self.addr}:{selfa.port}') # this will NEVER FLY
#--------------------------------------------^
self.scoket.connect(f'tcp://{self.addr}:{self.port}')  # this will ... ( + detect Error-state(s)

2)
Исправьте концепцию - смешивание низко висящих фруктов (несколько асинхронных каркасов, используемых одновременно, одна работает против всех остальных) признак поверхностного понимания ответственности профессионального инженера в надежных и достаточно надежных (fl aws в концепции управления, как при посадке Аполлона-11 на Луну или, наоборот, чернобыльская система неправильного управления + менталитет чисты, вдохновляют и достаточно примеров предупреждений о том, как плохие действия (если не запрещено + обнаружено + исправлено + наказано) может & снова и снова будет вредить ).

В случае, если кто-то никогда не работал с ZeroMQ, здесь можно с первого взгляда взглянуть на "Принципы ZeroMQ * менее чем за Пять секунд " перед тем, как углубляться в детали

Лучший следующий шаг

Если вы действительно стремитесь к профессиональному уровню, начните с книги Питера Хинтдженса "Code Connected, Том 1 »- стоит времени, стоит усилий, стоит разобраться в обсуждаемых здесь понятиях.

0 голосов
/ 17 апреля 2020 
#ADAPTED FROM PYMATA EXPRESS EXAMPLE CONCURRENTTAKS
#https://github.com/MrYsLab/pymata-express/
import asyncio
import zmq
import json
import zmq.asyncio as zmq_asyncio
from pymata_express.pymata_express import PymataExpress


class ConcurrentTasks:

    def __init__(self, board):


        self.loop = board.get_event_loop()
        self.board = board

        self.ctxsync = zmq.Context()
        self.context = zmq.asyncio.Context()
        self.rep = self.context.socket(zmq.REP)
        self.rep.bind("tcp://*:5558")

        self.trigger_pin = 53
        self.echo_pin = 51

        loop.run_until_complete(self.async_init_and_run())

    ### START:  NEW CODE THAT RESOLVED THE ISSUE
    async def pingsonar(self):
        value = await self.board.sonar_read(self.trigger_pin)
        return value

    async def readsonar(self):
        while True:
            rep_recv = await self.rep.recv() 
            value = await asyncio.wait([self.pingsonar()])
            valuesonar = list(value[0])[0].result()
            json_data = json.dumps(valuesonar) 
            await self.rep.send(json_data.encode()) 
            await asyncio.sleep(1 / 1000) #maybe this line isn't necessary

    ### END : NEW CODE THAT RESOLVED THE ISSUE

    async def async_init_and_run(self):

        await self.board.set_pin_mode_sonar(self.trigger_pin, self.echo_pin)

        readsonar = asyncio.create_task(self.readsonar())
        await readsonar

        # OTHER CREATED_TASK GO HERE, (removed them in the MVE, but they work fine)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    my_board = PymataExpress()
    try:
        ConcurrentTasks(my_board)
    except (KeyboardInterrupt, RuntimeError):
        loop.run_until_complete(my_board.shutdown())
        print('goodbye')
    finally:
        loop.close()
...