Orderer/connection.py

import datetime
import json
from typing import Dict, List, Any

import requests
from geventwebsocket import WebSocketError
from geventwebsocket.websocket import WebSocket

import model
from debug import debug
from my_types import MessageType, Message, UserIdentification, UserId, MessageQueue
from routes import push_message_types

PORT = 58317
if debug:
    host = 'http://localhost:' + str(PORT)
else:
    host = 'http://koljastrohm-games.com:' + str(PORT)

websockets_for_user: Dict[UserIdentification, List[WebSocket]] = {}
users_for_websocket: Dict[WebSocket, List[UserIdentification]] = {}
push_message_queue: MessageQueue = []
session_id = None


class HttpError(Exception):
    def __init__(self, code: int, message: Any = None, prepend_code=True):
        Exception.__init__(self)
        self.body = dict()
        self.code = code
        if message is None:
            message = str(code)

        if prepend_code and isinstance(message, str) and not message.startswith(str(code)):
            message = str(code) + ': ' + message
            if code // 100 == 2:
                self.body['message'] = message
            else:
                self.body['error'] = message
        elif isinstance(message, str):
            if code // 100 == 2:
                self.body['message'] = message
            else:
                self.body['error'] = message
        elif isinstance(message, dict):
            self.body = message.copy()
        else:  # for example a list or number
            self.body['data'] = message


class NotFound(HttpError):
    def __init__(self, message: Any = None, prepend_code=True):
        HttpError.__init__(self, 404, message, prepend_code)


class Success(HttpError):
    def __init__(self, message: Any = None, prepend_code=False):
        HttpError.__init__(self, 200, message, prepend_code)


class UnavailableForLegalReasons(HttpError):
    def __init__(self, message: Any = None, prepend_code=True):
        HttpError.__init__(self, 451, message, prepend_code)


class Forbidden(HttpError):
    def __init__(self, message: Any = None, prepend_code=True):
        HttpError.__init__(self, 403, message, prepend_code)


class Unauthorized(HttpError):
    def __init__(self, message: Any = None, prepend_code=True):
        HttpError.__init__(self, 401, message, prepend_code)


class BadRequest(HttpError):
    def __init__(self, message: Any = None, prepend_code=True):
        HttpError.__init__(self, 400, message, prepend_code)


class InternalServerError(HttpError):
    def __init__(self, message: Any = None, prepend_code=True):
        HttpError.__init__(self, 500, message, prepend_code)


class Locked(HttpError):
    def __init__(self, message: Any = None, prepend_code=True):
        HttpError.__init__(self, 423, message, prepend_code)


class PreconditionFailed(HttpError):
    def __init__(self, message: Any = None, prepend_code=True):
        HttpError.__init__(self, 412, message, prepend_code)


def check_missing_attributes(request_json: Dict, attributes: List[str]):
    for attr in attributes:
        if attr not in request_json:
            if str(attr) == 'session_id':
                raise Unauthorized('You are not signed in.')
            raise BadRequest('Missing value for attribute ' + str(attr))
        if str(attr) == 'session_id':
            if not model.valid_session_id(request_json['session_id']):
                raise Unauthorized('You are not signed in.')


def client_request(route, data=None):
    if data is None:
        data = {}
    return json_request(host + '/json/' + route, data)


def json_request(url, data):
    if debug:
        print('Sending to ' + url + ': ' + str(json.dumps(data)))
    r = requests.post(url,
                      data=json.dumps(data),
                      headers={'Content-type': 'application/json; charset=latin-1'})
    if debug:
        print('Request returned: ' + str(r.content))
    return r.json()


def push_message(recipient_ids: List[UserIdentification], contents: Message, message_type: MessageType):
    if message_type not in push_message_types:
        raise AssertionError('Invalid message type.')
    sockets = {socket for user_id in recipient_ids for socket in websockets_for_user.get(user_id, [])}
    if len(sockets) > 0:
        message = {'message_type': message_type, 'contents': contents}
        for ws in sockets:
            if ws.closed:
                ws_cleanup(ws)
                continue
            message = json.dumps({'message_type': message_type, 'contents': contents})
            ws.send(message)
        print(message_type,
              'to',
              len(sockets),
              'sockets',
              datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
              len(message))


def enqueue_push_message(recipient_ids: List[UserId], contents: Dict, message_type: str):
    assert message_type in push_message_types
    recipient_ids = [user_id for user_id in recipient_ids]
    if len(recipient_ids) == 0:
        return
    recipient_ids = [(model.current_db_name, user_id) for user_id in recipient_ids]
    push_message_queue.append((recipient_ids, contents, message_type))


def ws_cleanup(ws):
    if ws in users_for_websocket:
        users_affected = users_for_websocket[ws]
        for user_id in users_affected:
            # remove the websocket from all lists
            websockets_for_user[user_id][:] = filter(lambda s: s != ws,
                                                     websockets_for_user[user_id])
        del users_for_websocket[ws]
        if not ws.closed:
            ws.close()
    print('websocket connection ended',
          *ws.handler.client_address,
          datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), )


def preprocess_push_message_queue(queue: MessageQueue) -> MessageQueue:
    return queue


def push_messages_in_queue():
    global push_message_queue

    push_message_queue = preprocess_push_message_queue(push_message_queue)

    for message in push_message_queue:
        try:
            push_message(*message)
        except WebSocketError:
            continue
        except ConnectionResetError:
            continue
    del push_message_queue[:]