138ab4925d8c024c1eb7b95385207b0f43913e0f
[ais.git] / bin / inputs / outpeer.py
1 # -*- coding: utf-8 -*-
2 '''
3 UDP out peers module
4 '''
5
6 from __future__ import division
7 import logging
8 from time import time as get_timestamp, sleep
9 import socket
10 from threading import Thread, Lock
11
12 from ais.ntools import formataddr
13
14 __all__ = [ 'OutPeers', 'outpeers_from_config', 'TcpInOutPeer', 'TcpInServiceOut', 'tcpin_outpeers' ]
15
16 class UdpOutPeer:
17     '''
18     UDP output feed.
19     '''
20     def __init__(self, str_addrport):
21         colpos = str_addrport.rfind(':')
22         if colpos == 1:
23             assert colpos != 1, \
24                 'Missing column in host:port string for %s' % str_addrport
25         self.host = str_addrport[:colpos]
26         if self.host.startswith('['):
27             assert self.host[-1] == ']'
28             self.host = self.host[1:-1]
29         self.port = int(str_addrport[colpos+1:])
30         self.possibles_addrinfo = None
31         self.sock = None
32         self.addr = None
33         self.timestamp_getaddrinfo = 0
34         self._reinit_socket()
35
36     def __repr__(self):
37         return '%s:%s' % (self.host, self.port)
38
39     def _reinit_socket(self):
40         '''
41         Remove last known addrinfo for that host:port, if any.
42         If addrinfo is empty, fetch a new one.
43         Initialise self.sock and self.addr
44         '''
45         while True:
46             if self.possibles_addrinfo:
47                 # remove current (0th) addrinfo
48                 del self.possibles_addrinfo[0]
49
50             if not self.possibles_addrinfo:
51                 timestamp = get_timestamp()
52                 if timestamp > self.timestamp_getaddrinfo + 60:
53                     try:
54                         logging.debug('Calling socket.getaddrinfo for %s:%s',
55                             self.host, self.port)
56                         self.possibles_addrinfo = socket.getaddrinfo(self.host,
57                             self.port, socket.AF_UNSPEC, socket.SOCK_DGRAM)
58                         logging.info('%s possibles addrinfo for %s:%s.', 
59                             len(self.possibles_addrinfo), self.host, self.port)
60                         logging.debug('possibles_addrinfo: %s ',
61                             self.possibles_addrinfo)
62                     except socket.gaierror, err:
63                         logging.error("Can't resolve %s:%s: %s. "
64                             "Will retry in a minute.",
65                             self.host, self.port, err)
66                         return False
67                     finally:
68                         self.timestamp_getaddrinfo = timestamp
69                     if not self.possibles_addrinfo:
70                         logging.error("Can't resolve %s:%s. "
71                             "Will retry in a minute.",
72                             self.host, self.port)
73                         return False
74                 else:
75                     return False # waiting 
76
77             addrinfo = self.possibles_addrinfo[0]
78             logging.info('Prepared socket to %s', addrinfo)
79             af_, socktype, proto = addrinfo[:3]
80             self.addr = addrinfo[4]
81             try:
82                 self.sock = socket.socket(af_, socktype, proto)
83                 return True
84             except socket.error, msg:
85                 logging.error("Can't open socket to %s: %s", self.addr, msg)
86         
87     def send_line(self, line):
88         '''
89         Sends a AIVDM line to that peer.
90         CRLF will be added automatically.
91         '''
92         while self.sock or self._reinit_socket():
93             try:
94                 self.sock.sendto(line+'\r\n', self.addr)
95                 return
96             except socket.error, err:
97                 logging.error('socket.sendto(%s) failed: %s', self.addr, err)
98                 self.sock = None
99
100
101 class TcpInOutPeer:
102     def __init__(self, conn, addr, port):
103         self.sock = conn
104         self.sock.setblocking(0)
105         self.addr = addr
106         self.port = port
107         #logging.debug('Set linger time for outputtcpinlistener socket to %s seconds', SERVER_LINGER_TIME)
108         #self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
109         #                      struct.pack('ii', 1, SERVER_LINGER_TIME))
110
111     def send_line(self, line):
112         try:
113             self.sock.send(line+'\r\n')
114         except socket.error, err:
115             if err.errno == 32: # Broken pipe
116                 logging.info('TcpInOutPeer %s:%s disconnect', formataddr(self.addr), self.port)
117                 tcpin_outpeers.remove_peer(self)
118             else:
119                 logging.error('socket.send() to TcpInOutPeer %s:%s failed %s', formataddr(self.addr), port, repr(err))
120
121     def shutdown(self):
122         logging.info('Shuting down TcpInOutPeer socket %s:%s', formataddr(self.addr), self.port)
123         self.sock.setblocking(1)
124         self.sock.shutdown(socket.SHUT_RDWR)
125
126
127
128 class TcpInServiceOut(Thread):
129     def __init__(self, port):
130         Thread.__init__(self)
131         self.port = port
132         self.sock = None
133
134     def run(self):
135         self.sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
136         self.sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
137         while True:
138             try:
139                 self.sock.bind(('', self.port))
140                 break
141             except socket.error, err:
142                 if err.errno == 98: # Address already in use
143                     logging.error('TCP port %s already in use. Waiting...', self.port)
144                     sleep(1)
145                 else:
146                     raise
147         logging.info('Bound to socket on port %s for listeners', self.port)
148         #logging.debug('Set linger time for tcpinoutpeerlisten socket to %s seconds', SERVER_LINGER_TIME)
149         #self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
150         #                      struct.pack('ii', 1, SERVER_LINGER_TIME))
151         self.sock.listen(2)
152         while True:
153             try:
154                 conn, addr = self.sock.accept()
155             except socket.error, err:
156                 if err.errno == 22: # Invalid argument
157                     logging.info('TcpInListen socket was closed. Exiting thread.')
158                     return
159                 raise
160             logging.info('Incoming TcpInOutPeer %s:%s', formataddr(addr[0]), addr[1])
161             tcpin_outpeers.add_peer(TcpInOutPeer(conn, addr[0], addr[1]))
162     
163     def shutdown(self):
164         if self.sock:
165             logging.info('Shuting down tcpinoutpeerlisten socket')
166             self.sock.shutdown(socket.SHUT_RDWR)
167         
168
169 class OutPeers:
170     '''
171     A group of UDP peers for output.
172     This class is thread safe.
173     '''
174     def __init__(self):
175         self._peers = set()
176         self._lock = Lock()
177
178     def __repr__(self):
179         return '(' + ', '.join([ repr(peer) for peer in self._peers]) + ')'
180         
181     def add_peer(self, peer):
182         self._lock.acquire()
183         self._peers.add(peer)
184         self._lock.release()
185
186     def remove_peer(self, peer):
187         self._lock.acquire()
188         self._peers.remove(peer)
189         self._lock.release()
190
191     def safe_iter(self):
192         # make a copy of the set, because its size may change during iteration (RuntimeError)
193         self._lock.acquire()
194         tmp_peers = [ peer for peer in self._peers ]
195         self._lock.release()
196         for peer in tmp_peers:
197             yield peer
198
199     def send_line(self, line):
200         '''
201         Send a line to all the output peers registered.
202         '''
203         for outpeer in self.safe_iter():
204             logging.debug('OUT %s %s', outpeer, repr(line))
205             outpeer.send_line(line)
206
207 def outpeers_from_config(list_from_config):
208     '''
209     Just give the list of strings from config.
210     Returns an OutPeers object.
211     '''
212     outpeers = OutPeers()
213     for str_config in list_from_config:
214         assert str_config.startswith('udp:'), \
215                'Unsupported output peer protocol %s' % str_config
216         peer = UdpOutPeer(str_config[4:])
217         outpeers.add_peer(peer)
218     return outpeers
219
220
221 tcpin_outpeers = OutPeers()