janus_demo_1

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import janus
import asyncio
import time
import threading
from bson.objectid import ObjectId
from autobahn.asyncio.wamp import ApplicationSession
from autobahn.asyncio.wamp import ApplicationRunner

queue = janus.Queue()
squeue = janus.Queue()

_dict = {}


async def y2():
    while True:
        result = await queue.async_q.get()
        _dict[result[0]].set_result(result[1])


def t3(*args, **kwargs):
    time.sleep(2)
    return [1111, 1244]


def asynchronous_code():
    while True:
        [tid, fut] = squeue.sync_q.get()
        result = fun
        queue.sync_q.put([tid, result])

            
class FrontendSession(ApplicationSession):

    def onConnect(self):
        self.join(self.config.realm, [u"ticket"], u'backend')

    def onChallenge(self, challenge):
        if challenge.method == u"ticket":
            return u'sg-ai.com'
        else:
            raise Exception("Invalid authmethod {}".format(challenge.method))

    async def onJoin(self, details):
        print(self._session_id)
        asyncio.ensure_future(y2())
        async def start1():
            tid = str(ObjectId())
            # !!! asyncio.exsure_future(), can only execute asynchronous functions 
            fut = asyncio.ensure_future(t3())
            await squeue.async_q.put([tid, fut])
            _dict[tid] = fut
            while 1:
                try:
                    # After executing this line of code, the task will be executed directly,
                    # but the result will not be available immediately. When using await to parse the return,
                    # it is possible to get the result.
                    result = await fut.result()
                    _dict.pop(tid)
                    return result
                except Exception as e:
                    # There is an error reading the result directly before the result is obtained.
                    pass
                # The experiment shows that the following method will permanently pause this loop here.
                # But if you don't use it, you can't run the task later.
                await asyncio.sleep(0.5)

        st = time.time()
        res = await start1()
        print(res)
        print('%fs' % (time.time() - st))

    def onLeave(self, details):
        print("Client session left: {}".format(details))
        self.disconnect()

    def onDisconnect(self):
        print("Client session disconnected.")


if __name__ == '__main__':
    t = threading.Thread(target=asynchronous_code)
    t.setDaemon(True)
    t.start()

    runner = ApplicationRunner(url='wss://dcrossbar.sg-ai.com/ws', realm='realm1')
    runner.run(FrontendSession, log_level='debug')