Ожидание как gevent socket, так и redis blocking blpop

Question

Я использую gevent StreamServer для обработки входящих соединений с клиентами.

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

Я сделал бы это с redis. Я создал очередь с определенным идентификатором клиента в качестве ключа. После того, как клиент отправит сообщение, я проверяю очередь и, если есть какое-либо сообщение, я отправляю его клиенту.

Недостатком этого подхода является то, что сервер может отправлять сообщение только после того, как клиент отправит сообщение.

Есть ли способ, чтобы я мог ждать входящих данных и redis blpop, поэтому я могу отправить сообщение обратно клиенту, как только сообщение будет готово, вместо того, чтобы ждать, пока клиент отправит следующие данные?

import gevent 
from gevent import socket 
from gevent.server import StreamServer 
import redis 

r = redis.Redis('localhost') 

def handle_echo(sock, address): 
    fp = sock.makefile() 
    while True: 
     line = fp.readline() 
     if line: 
      client_id = line.split(",")[0] 
      if r.llen('%s:servercmds' % client_id) > 0: 
       tosend = r.lrange('%s:servercmds' % imei, 0, 0)[0] 
       try:   
        fp.write(tosend) 
        fp.flush() 
        r.lpop('%s:servercmds' % imei) 
       except: 
        print('cannot send data to client') 
      else: 
       break 
     sock.shutdown(socket.SHUT_WR) 
     sock.close() 

server = StreamServer(('', 8045), handle_echo, spawn=10000) 
server.serve_forever() 

Answer 1

Вам необходимо, чтобы клиент открыл соединение, чтобы вы могли отправить ему сообщение или прослушать сообщения.

Редактировать: это код с верхней части головы. ИМО обычно разделяет чтение и запись таким образом. Однако это не единственный шаблон, который вы можете использовать.

import gevent 
from gevent import socket 
from gevent.server import StreamServer 
import redis 

r = redis.Redis('localhost') 

def handle_echo(sock, address): 
    def read_loop(sock): 
     while True: 
      try: 
       socket.wait_read(sock.fileno()) 
      except socket.error: 
       break 
      # read from socket 

    def write_loop(sock): 
     while True: 
      try: 
       socket.wait_write(sock.fileno()) 
      except socket.error: 
       break 
      # write to socket 

    jobs = [gevent.spawn(func, sock) for func in (read_loop, write_loop)] 
    gevent.joinall(jobs) 

server = StreamServer(('', 8045), handle_echo) 
server.serve_forever()