janus_demo

#!/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 = {}


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


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


def asynchronous_code():
    while True:
        t_info = squeue.sync_q.get()
        for tid, fut in t_info.items():
            result = fut.run()
            queue.sync_q.put([tid, result])


class Future(object):
    def __init__(self, func, *args, **kwargs):
        self._result = None
        self.func = func
        self.args = args
        self.kwargs = kwargs

    async def result(self):
        while True:
            if self._result is not None:
                return self._result
            await asyncio.sleep(0.5)

    def set_result(self, value):
        self._result = value

    def run(self):
        return self.func(*self.args, **self.kwargs)


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)

        async def start1():
            tid = str(ObjectId())
            fut = Future(t3)
            await squeue.async_q.put({tid: fut})
            _dict[tid] = fut
            return await fut.result()

        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)
    t1 = threading.Thread(target=y2)
    t.setDaemon(True)
    t1.setDaemon(True)
    t.start()
    t1.start()

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

    """
    //// results of enforcement ////
    """
    """
    
    2018-10-31T21:24:20 received HTTP response:

    b'HTTP/1.1 101 Switching Protocols\r\nServer: nginx\r\nDate: Wed, 31 Oct 2018 13:24:20 GMT\r\nConnection: upgrade\r\nUpgrade: WebSocket\r\nSec-WebSocket-Protocol: wamp.2.json.batched\r\nSec-WebSocket-Accept: wR58GMXY5FT7t56ccuhO8dFBVUA=\r\n\r\n'
    
    
    2018-10-31T21:24:20 received HTTP status line in opening handshake : HTTP/1.1 101 Switching Protocols
    2018-10-31T21:24:20 received HTTP headers in opening handshake : {'server': 'nginx', 'date': 'Wed, 31 Oct 2018 13:24:20 GMT', 'connection': 'upgrade', 'upgrade': 'WebSocket', 'sec-websocket-protocol': 'wamp.2.json.batched', 'sec-websocket-accept': 'wR58GMXY5FT7t56ccuhO8dFBVUA='}
    2018-10-31T21:24:20 openHandshakeTimeoutCall.cancel
    2130178912806293
    [1111, 1244]
    2.503509s
    2018-10-31T21:24:29 Auto ping/pong: sending ping auto-ping/pong
    2018-10-31T21:24:29 Expecting ping in 5.0 seconds for auto-ping/pong
    2018-10-31T21:24:29 Auto ping/pong: received pending pong for auto-ping/pong
    2018-10-31T21:24:29 WebSocketProtocol.onPong(payload=<4 bytes>)
    Client session left: CloseDetails(reason=<wamp.close.normal>, message='None')
    2018-10-31T21:24:31 connection closed properly: canceling closing handshake timeout
    2018-10-31T21:24:31 _connectionLost: None
    2018-10-31T21:24:31 serverConnectionDropTimeoutCall.cancel
    2018-10-31T21:24:31 Auto ping/pong: canceling autoPingPendingCall upon lost connection
    2018-10-31T21:24:31 WAMP-over-WebSocket transport lost: wasClean=True, code=1000, reason="None"
    Client session disconnected.
    """