condure 1.10.0

HTTP/WebSocket connection manager
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
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157

# this handler holds all connections open

import os
import time
import datetime
import calendar
import tnetstring
import zmq

CONN_TTL = 60000
EXPIRE_INTERVAL = 60000

instance_id = 'holdhandler.{}'.format(os.getpid()).encode('utf-8')

ctx = zmq.Context()
in_sock = ctx.socket(zmq.PULL)
in_sock.connect('ipc://client-out')
in_stream_sock = ctx.socket(zmq.ROUTER)
in_stream_sock.identity = instance_id
in_stream_sock.connect('ipc://client-out-stream')
out_sock = ctx.socket(zmq.PUB)
out_sock.connect('ipc://client-in')

poller = zmq.Poller()
poller.register(in_sock, zmq.POLLIN)
poller.register(in_stream_sock, zmq.POLLIN)

class Connection(object):
    def __init__(self, rid):
        self.rid = rid
        self.seq = 0
        self.exp_time = None

    def send_msg(self, msg):
        msg[b'from'] = instance_id
        msg[b'id'] = self.rid[1]
        msg[b'seq'] = self.seq
        self.seq += 1

        print('OUT {} {}'.format(self.rid[0], msg))
        out_sock.send(self.rid[0] + b' T' + tnetstring.dumps(msg))

    def send_header(self):
        msg = {}
        msg[b'code'] = 200
        msg[b'reason'] = b'OK'
        msg[b'headers'] = [[b'Content-Type', b'text/plain']]
        msg[b'more'] = True

        self.send_msg(msg)

    def send_body(self, data):
        msg = {}
        msg[b'body'] = data
        msg[b'more'] = True

        self.send_msg(msg)

def send_body(to_addr, conns, data):
    ids = []
    for c in conns:
        ids.append({b'id': c.rid[1], b'seq': c.seq})
        c.seq += 1

    msg = {}
    msg[b'from'] = instance_id
    msg[b'id'] = ids
    msg[b'body'] = data
    msg[b'more'] = True

    print('OUT {} {}'.format(to_addr, msg))
    out_sock.send(to_addr + b' T' + tnetstring.dumps(msg))

conns = {}
last_exp_time = int(time.time())

while True:
    socks = dict(poller.poll(1000))

    if socks.get(in_sock) == zmq.POLLIN:
        m_raw = in_sock.recv()
    elif socks.get(in_stream_sock) == zmq.POLLIN:
        m_list = in_stream_sock.recv_multipart()
        m_raw = m_list[2]
    else:
        m_raw = None

    now = int(time.time() * 1000)

    if m_raw is not None:
        req = tnetstring.loads(m_raw[1:])
        print('IN {}'.format(req))

        m_from = req[b'from']
        m_id = req[b'id']
        m_type = req.get(b'type', b'')

        ids = []
        if isinstance(m_id, list):
            for id_seq in m_id:
                ids.append(id_seq[b'id'])
        else:
            ids.append(m_id)

        new_ids = []
        known_conns = []
        for i in ids:
            rid = (m_from, i)

            c = conns.get(rid)
            if c:
                c.exp_time = now + CONN_TTL
                known_conns.append(c)
            else:
                new_ids.append(rid)

        # data
        if not m_type:
            for rid in new_ids:
                c = Connection(rid)
                conns[rid] = c
                c.exp_time = now + CONN_TTL
                c.send_header()
        elif c:
            if m_type == b'keep-alive':
                dt = datetime.datetime.utcnow()
                ts = calendar.timegm(dt.timetuple())

                body = (
                    'id: TCPKaliMsgTS-{:016x}.\n'
                    'event: message\n'
                    'data: {:04}-{:02}-{:02}T{:02}:{:02}:{:02}\n\n'
                ).format(
                    (ts * 1000000) + dt.microsecond,
                    dt.year,
                    dt.month,
                    dt.day,
                    dt.hour,
                    dt.minute,
                    dt.second
                ).encode()

                send_body(m_from, known_conns, body)
            elif m_type == b'cancel':
                for c in known_conns:
                    del conns[c.rid]

    if now >= last_exp_time + EXPIRE_INTERVAL:
        last_exp_time = now

        to_remove = []
        for rid, c in conns.items():
            if now >= c.exp_time:
                to_remove.append(rid)
        for rid in to_remove:
            print('expired {}'.format(rid))
            del conns[rid]