-
Notifications
You must be signed in to change notification settings - Fork 12
/
stationd
130 lines (105 loc) · 4.2 KB
/
stationd
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
#!/usr/bin/env python
import os
import sys
import select
import time
from groundstation import logger
log = logger.getLogger("stationd")
from groundstation import settings
from groundstation.station import Station
from groundstation.broadcast_discoverer import BroadcastDiscoverer
from groundstation.broadcast_announcer import BroadcastAnnouncer
from groundstation.stream_listener import StreamListener
from groundstation.stream_client import StreamClient
from groundstation.node import Node
from groundstation.peer_socket import PeerSocket
from groundstation.sockets.socket_closed_exception import SocketClosedException
from groundstation.peer_socket_pool import PeerSocketPool
from groundstation.broadcast_events import new_broadcast_event, UnknownBroadcastEvent
from groundstation.broadcast_events.broadcast_ping import BroadcastPing
from groundstation.gizmo_factory import InvalidGizmoError
from os.path import expanduser
PORT = settings.PORT
BEACON_TIMEOUT = settings.BEACON_TIMEOUT
myself = Node()
station_path = expanduser("~/.groundstation")
station = Station(station_path, myself)
discoverer = BroadcastDiscoverer(PORT)
announcer = BroadcastAnnouncer(PORT)
listener = StreamListener(PORT)
announcer.name = myself.name
sockets = [discoverer, listener]
peer_sockets = PeerSocketPool()
last_beacon = time.time() - BEACON_TIMEOUT # Gaurantee that we'll announce on the first run
def _read_sockets():
read_sockets = []
for i in sockets: read_sockets.append(i)
for i in peer_sockets: read_sockets.append(i)
return read_sockets
def _write_sockets():
write_sockets = []
for i in peer_sockets:
if i.has_data_ready():
write_sockets.append(i)
return write_sockets
def handle_discoverer_event():
# TODO Return Event objects instead of raw data
data, peer = discoverer.recv(settings.DEFAULT_BUFSIZE)
try:
event = new_broadcast_event(data)
except UnknownBroadcastEvent:
log.warning("Ooops got some weird event: %s" % (repr(data)))
if isinstance(event, BroadcastPing):
if event.payload == myself.uuid:
log.info("Discarding PING from myself")
else:
log.info("Ping from %s" % str(peer))
if event.payload > myself.uuid:
# Peer's uuid is larger, we should connect and initiate sync
if peer[0] not in peer_sockets:
# Ensure that they're in our object cache- we're updating it now
if not station.recently_queried(event.payload):
client = StreamClient(peer[0])
peer_sockets.append(client)
client.begin_handshake(station)
else:
# Peer's uuid is smaller, we should do nothing and await connection
log.info("Peer's uuid is smaller, awaiting connection")
def handle_listener_event():
peer = listener.accept(PeerSocket)
peer_sockets.append(peer)
def handle_tcpnetwork_event():
try:
i.recv()
while i.packet_queue:
data = i.packet_queue.pop()
gizmo = station.gizmo_factory.hydrate(data, i)
assert gizmo is not None, "gizmo_factory returned None"
gizmo.process()
except SocketClosedException:
peer_sockets.remove(i)
except InvalidGizmoError:
log.warn("Recieved invalid gizmo!")
def handler_for(thing):
if thing == discoverer:
return handle_discoverer_event
elif thing == listener:
return handle_listener_event
elif isinstance(thing, PeerSocket) or isinstance(i, StreamClient):
return handle_tcpnetwork_event
else:
raise KeyError
while True:
if time.time() > (last_beacon + BEACON_TIMEOUT):
last_beacon = time.time()
announcer.ping()
log.info("Blocking on sockets for up to %is" % (BEACON_TIMEOUT))
(sread, swrite, sexc) = select.select(_read_sockets(), _write_sockets(), [], BEACON_TIMEOUT)
log.info("Got %i read fds" % (len(sread)))
log.info("Got %i write fds" % (len(swrite)))
for i in sread:
handler_for(i)()
for i in swrite:
if isinstance(i, PeerSocket) or isinstance(i, StreamClient):
i.send()
# else: # Right now that's the only socket