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()