rostrum 14.0.1

An efficient implementation of Electrum Server with token support
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

import asyncio
import json
from typing import Any, Optional
import http.client
import ssl


def request(host, port, data, use_ssl=False, ssl_context=None):
    if use_ssl:
        conn = http.client.HTTPSConnection(host, port, timeout=10, context=ssl_context)
    else:
        conn = http.client.HTTPConnection(host, port, timeout=10)

    # Convert data to JSON string
    json_data = json.dumps(data)
    headers = {"Content-Type": "application/json"}

    print(f"{'https' if use_ssl else 'http'}: Sending request {host}:{port}", json_data)
    conn.request("POST", "/", body=json_data, headers=headers)
    resp = conn.getresponse()
    return json.loads(resp.read().decode())


# pylint: disable=too-many-boolean-expressions,too-many-positional-arguments,too-many-arguments
async def request_async(loop, host, port, data, use_ssl=False, ssl_context=None):
    def send_request():
        return request(host, port, data, use_ssl, ssl_context)

    # Offload the blocking I/O operation to a thread pool
    return await loop.run_in_executor(None, send_request)


# pylint: disable=no-member,protected-access
class HTTPProtocol:
    client: Any = None  # circluar reference, cant define type
    host: str
    loop: asyncio.AbstractEventLoop
    port: int
    use_ssl: bool
    ssl_context: Optional[ssl.SSLContext]

    # pylint: disable=too-many-arguments,too-many-positional-arguments
    def __init__(
        self,
        host: str,
        port: int,
        client,
        loop: asyncio.AbstractEventLoop,
        use_ssl=False,
        ssl_context=None,
    ):
        self.client = client
        self.host = host
        self.loop = loop
        self.port = port
        self.use_ssl = use_ssl
        self.ssl_context = ssl_context

    async def connect(self):
        # Send a ping to see if it's possible to connect (throws on connection error)
        request(
            self.host,
            self.port,
            {"method": "server.ping", "params": []},
            self.use_ssl,
            self.ssl_context,
        )

    async def send_data_task(self, data: dict):
        try:
            response = await request_async(
                self.loop, self.host, self.port, data, self.use_ssl, self.ssl_context
            )
            self.client._got_response(response)

        except (http.client.HTTPException, ConnectionError) as e:
            print("HTTP request failed:", e)
            self.client.connection_lost(e)
        except Exception as e:
            print("Unexpected error occurred:", e)
            self.client.connection_lost(e)

    def send_data(self, data: dict):
        self.loop.create_task(self.send_data_task(data))

    def close(self):
        pass