using SignalR in python

Question

import asyncio

from signalr_async.netcore import Hub, Client

import uuid

import socket

import os

import requests

 

BASEURL = "http://localhost:5007/"

class MyHub(Hub):

    async def on_connect(self, connection_id: str) -> None:

       print(f'Connected Deviced: {connection_id}')

    async def on_disconnect(self) -> None:

         print('Disconnected')

       

    def on_CountSwine(self, CageId: str) -> None:

        print(f"Cage id received: {CageId}")

        if len(CageId) > 0:

           print(f"do some things")

 

hub = MyHub("applicationHub")

async def main():  

    deviceId = str(uuid.uuid1())

    ipLan = socket.gethostbyname(socket.gethostname())

    osName = os.name

    headers = {"deviceId": deviceId, "ip": ipLan, "os": osName}

    client = Client(

        BASEURL,

        hub,

        connection_options={        

            "ws_client_options": {"headers": headers, "timeout": 1.0},          

        },

    )

    await client.start()  

loop = asyncio.get_event_loop()

loop.create_task(main())

loop.run_forever()

// lưu lại khi nào cần xài thì xài

Answer 1

Đã test ok và sử dụng ngon lành cành đào