Настройка Apache Kafka Consumer соединения внутри Django канала? - PullRequest
Новая
       22

Настройка Apache Kafka Consumer соединения внутри Django канала?

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

Надеюсь, у всех все хорошо.

У меня есть вопрос по настройке потребителя канала django. Я знаю, что есть возможность подключить javascript websocket к django потребителю канала. Я уже подключил эту часть, но знаю, что когда соединение будет установлено, я бы хотел запустить потребителя кафки. Я достиг этого также. Однако, как только WebSocket отключается, потребитель канала django никогда не отключается полностью, так как там он занят.

У меня вопрос, возможно ли выполнить sh такую ​​вещь без использования какого-либо канального слоя и повторного отображения?

Сообщение об ошибке после отключения соединения 

Application instance <Task pending coro=<SessionMiddlewareInstance.__call__() running at /home/bd/anaconda2/lib/python3.6/site-packages/channels/sessions.py:183> wait_for=<Future pending cb=[_chain_future.<locals>._call_check_cancel() at /home/bd/anaconda2/lib/python3.6/asyncio/futures.py:407, <TaskWakeupMethWrapper object at 0x7f2a3ee78228>()]>> for connection <WebSocketProtocol client=['127.0.0.1', 53850] path=b'/'> took too long to shut down and was killed.

Javascript: 

....
const socket = new WebSocket("ws://")

socket.onmessage = function(e){
        console.log("onmessage",e)
    }

socket.onopen = function(e){
        console.log("onopen",e)
    }
....

Django Потребитель канала: 

class Test(WebSocketConsumer):
...
def connect():
   self.acccept()

   c = Consumer({
            'bootstrap.servers': 'localhost:9092',
            'group.id': "sample"
       })

   c.subscribe(['test'])

   while True:
     msg = c.poll() 

     ....
     some conditions
     ...
     if msg.key().decode("utf-8") == "someText":
        text = msg.value().decode("utf-8")
        self.send(text)

  c.close()
...