scythargon/asyncio_server.py

## asyncio_server.py
"""
As a client can use the next html page:

<!DOCTYPE html>
<html>
<head>
<script type="text/javascript">
    window.addEventListener("load", function() {
        // create websocket instance
        var mySocket = new WebSocket("ws://localhost:8080/ws");
        // add event listener reacting when message is received
        mySocket.onmessage = function (event) {
            var output = document.getElementById("output");
            // put text into our output div
            output.textContent = event.data;
        };
        var form = document.getElementsByClassName("foo");
        var input = document.getElementById("input");
        form[0].addEventListener("submit", function (e) {
            // on forms submission send input to our server
            input_text = input.value;
            mySocket.send(input_text);
            e.preventDefault()
        })
    });
</script>
<style>
    /* just some super ugly css to make things bit more readable*/
    div {
        margin: 10em;
    }
    form {
        margin: 10em;
    }
</style>
</head>
<body>
    <form class="foo">
        <input id="input"></input>
        <input type="submit"></input>
    </form>
    <div id="output"></div>
</body>
</html>

"""

import asyncio
import os
import threading
from slack import RTMClient, WebClient
from slack.errors import SlackApiError

import random

from autobahn.asyncio.websocket import (
    WebSocketServerProtocol,
    WebSocketServerFactory
)

slack_client = WebClient(
    token=os.environ['SLACK_API_TOKEN']
)


@RTMClient.run_on(event='message')
def say_hello(**payload):
    data = payload['data']
    web_client = payload['web_client']
    rtm_client = payload['rtm_client']
    text = data.get('text', '')
    # print(data)
    if 'Hello' in text:
        # print('Received message')
        channel_id = data['channel']
        thread_ts = data['ts']
        user = data['user']

        try:
            rtm_client.factory.broadcast(f"New message from slack: {text}")
            rtm_client.factory.last_channel = channel_id
            rtm_client.factory.last_thread_ts = thread_ts

            response = web_client.chat_postMessage(
                channel=channel_id,
                text=f"Hi <@{user}>!",
                thread_ts=thread_ts
            )
        except SlackApiError as e:
            # You will get a SlackApiError if "ok" is False
            assert e.response["ok"] is False
            assert e.response["error"]  # str like 'invalid_auth', 'channel_not_found'
            print(f"Got an error: {e.response['error']}")


def run_slack_client(factory):
    second_loop = asyncio.new_event_loop()
    asyncio.set_event_loop(second_loop)
    rtm_client = RTMClient(token=os.environ["SLACK_API_TOKEN"])
    rtm_client.factory = factory
    rtm_client.start()


class SomeServerProtocol(WebSocketServerProtocol):
    def onOpen(self):
        """
        Connection from client is opened. Fires after opening
        websockets handshake has been completed and we can send
        and receive messages.

        Register client in factory, so that it is able to track it.
        Try to find conversation partner for this client.
        """
        self.factory.register(self)
        self.factory.findPartner(self)

    def connectionLost(self, reason):
        """
        Client lost connection, either disconnected or some error.
        Remove client from list of tracked connections.
        """
        self.factory.unregister(self)

    def onMessage(self, payload, isBinary):
        """
        Message sent from client, communicate this message to its conversation partner,
        """
        self.factory.communicate(self, payload, isBinary)


class ChatRouletteFactory(WebSocketServerFactory):
    def __init__(self, *args, **kwargs):
        super(ChatRouletteFactory, self).__init__(*args, **kwargs)
        self.clients = {}

    def register(self, client):
        """
        Add client to list of managed connections.
        """
        self.clients[client.peer] = {"object": client, "partner": None}

    def unregister(self, client):
        """
        Remove client from list of managed connections.
        """
        self.clients.pop(client.peer)

    def findPartner(self, client):
        """
        Find chat partner for a client. Check if there any of tracked clients
        is idle. If there is no idle client just exit quietly. If there is
        available partner assign him/her to our client.
        """
        available_partners = [c for c in self.clients if c != client.peer and not self.clients[c]["partner"]]
        if not available_partners:
            print("no partners for {} check in a moment".format(client.peer))
        else:
            partner_key = random.choice(available_partners)
            self.clients[partner_key]["partner"] = client
            self.clients[client.peer]["partner"] = self.clients[partner_key]["object"]

    def communicate(self, client, payload, isBinary):
        """
        Broker message from client to its partner.
        """
        c = self.clients[client.peer]
        if self.last_channel and self.last_thread_ts:
            slack_client.chat_postMessage(
                channel=self.last_channel,
                text=f"Message from chat: {payload.decode('utf-8')}",
                thread_ts=self.last_thread_ts
            )

        if not c["partner"]:
            c["object"].sendMessage("Sorry you dont have partner yet, check back in a minute".encode('utf-8'))
        else:
            c["partner"].sendMessage(payload)

    def broadcast(self, message):
        """Send a slack message to all connected clients."""
        print('broadcast call')
        for client in self.clients.values():
            client["object"].sendMessage(message.encode('utf-8'))


if __name__ == '__main__':
    factory = ChatRouletteFactory("ws://127.0.0.1:9000")
    factory.protocol = SomeServerProtocol

    loop = asyncio.get_event_loop()

    thread = threading.Thread(target=run_slack_client, args=(factory,))
    thread.daemon = True
    thread.start()

    coro = loop.create_server(factory, '0.0.0.0', 9000)
    server = loop.run_until_complete(coro)

    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.close()
        loop.close()
	"""
	As a client can use the next html page:

	<!DOCTYPE html>
	<html>
	<head>
	<script type="text/javascript">
	window.addEventListener("load", function() {
	// create websocket instance
	var mySocket = new WebSocket("ws://localhost:8080/ws");
	// add event listener reacting when message is received
	mySocket.onmessage = function (event) {
	var output = document.getElementById("output");
	// put text into our output div
	output.textContent = event.data;
	};
	var form = document.getElementsByClassName("foo");
	var input = document.getElementById("input");
	form[0].addEventListener("submit", function (e) {
	// on forms submission send input to our server
	input_text = input.value;
	mySocket.send(input_text);
	e.preventDefault()
	})
	});
	</script>
	<style>
	/* just some super ugly css to make things bit more readable*/
	div {
	margin: 10em;
	}
	form {
	margin: 10em;
	}
	</style>
	</head>
	<body>
	<form class="foo">
	<input id="input"></input>
	<input type="submit"></input>
	</form>
	<div id="output"></div>
	</body>
	</html>

	"""

	import asyncio
	import os
	import threading
	from slack import RTMClient, WebClient
	from slack.errors import SlackApiError

	import random

	from autobahn.asyncio.websocket import (
	WebSocketServerProtocol,
	WebSocketServerFactory
	)

	slack_client = WebClient(
	token=os.environ['SLACK_API_TOKEN']
	)


	@RTMClient.run_on(event='message')
	def say_hello(**payload):
	data = payload['data']
	web_client = payload['web_client']
	rtm_client = payload['rtm_client']
	text = data.get('text', '')
	# print(data)
	if 'Hello' in text:
	# print('Received message')
	channel_id = data['channel']
	thread_ts = data['ts']
	user = data['user']

	try:
	rtm_client.factory.broadcast(f"New message from slack: {text}")
	rtm_client.factory.last_channel = channel_id
	rtm_client.factory.last_thread_ts = thread_ts

	response = web_client.chat_postMessage(
	channel=channel_id,
	text=f"Hi <@{user}>!",
	thread_ts=thread_ts
	)
	except SlackApiError as e:
	# You will get a SlackApiError if "ok" is False
	assert e.response["ok"] is False
	assert e.response["error"] # str like 'invalid_auth', 'channel_not_found'
	print(f"Got an error: {e.response['error']}")


	def run_slack_client(factory):
	second_loop = asyncio.new_event_loop()
	asyncio.set_event_loop(second_loop)
	rtm_client = RTMClient(token=os.environ["SLACK_API_TOKEN"])
	rtm_client.factory = factory
	rtm_client.start()


	class SomeServerProtocol(WebSocketServerProtocol):
	def onOpen(self):
	"""
	Connection from client is opened. Fires after opening
	websockets handshake has been completed and we can send
	and receive messages.

	Register client in factory, so that it is able to track it.
	Try to find conversation partner for this client.
	"""
	self.factory.register(self)
	self.factory.findPartner(self)

	def connectionLost(self, reason):
	"""
	Client lost connection, either disconnected or some error.
	Remove client from list of tracked connections.
	"""
	self.factory.unregister(self)

	def onMessage(self, payload, isBinary):
	"""
	Message sent from client, communicate this message to its conversation partner,
	"""
	self.factory.communicate(self, payload, isBinary)


	class ChatRouletteFactory(WebSocketServerFactory):
	def __init__(self, args, *kwargs):
	super(ChatRouletteFactory, self).__init__(args, *kwargs)
	self.clients = {}

	def register(self, client):
	"""
	Add client to list of managed connections.
	"""
	self.clients[client.peer] = {"object": client, "partner": None}

	def unregister(self, client):
	"""
	Remove client from list of managed connections.
	"""
	self.clients.pop(client.peer)

	def findPartner(self, client):
	"""
	Find chat partner for a client. Check if there any of tracked clients
	is idle. If there is no idle client just exit quietly. If there is
	available partner assign him/her to our client.
	"""
	available_partners = [c for c in self.clients if c != client.peer and not self.clients[c]["partner"]]
	if not available_partners:
	print("no partners for {} check in a moment".format(client.peer))
	else:
	partner_key = random.choice(available_partners)
	self.clients[partner_key]["partner"] = client
	self.clients[client.peer]["partner"] = self.clients[partner_key]["object"]

	def communicate(self, client, payload, isBinary):
	"""
	Broker message from client to its partner.
	"""
	c = self.clients[client.peer]
	if self.last_channel and self.last_thread_ts:
	slack_client.chat_postMessage(
	channel=self.last_channel,
	text=f"Message from chat: {payload.decode('utf-8')}",
	thread_ts=self.last_thread_ts
	)

	if not c["partner"]:
	c["object"].sendMessage("Sorry you dont have partner yet, check back in a minute".encode('utf-8'))
	else:
	c["partner"].sendMessage(payload)

	def broadcast(self, message):
	"""Send a slack message to all connected clients."""
	print('broadcast call')
	for client in self.clients.values():
	client["object"].sendMessage(message.encode('utf-8'))


	if __name__ == '__main__':
	factory = ChatRouletteFactory("ws://127.0.0.1:9000")
	factory.protocol = SomeServerProtocol

	loop = asyncio.get_event_loop()

	thread = threading.Thread(target=run_slack_client, args=(factory,))
	thread.daemon = True
	thread.start()

	coro = loop.create_server(factory, '0.0.0.0', 9000)
	server = loop.run_until_complete(coro)

	try:
	loop.run_forever()
	except KeyboardInterrupt:
	pass
	finally:
	server.close()
	loop.close()