bin/inputs/tcpout.py

   1 #!/usr/bin/env python
   2
   3 '''
   4 Module for receiving AIVDM data from outbound TCP connection.
   5 '''
   6
   7 from __future__ import division
   8 import logging
   9 import socket
  10
  11 from ais.inputs.common import DEFAULT_MTU, get_source_by_id4
  12 from ais.inputs.virtual import Channel, ServiceIn
  13
  14 TCP_HEADER_SIZE = 52
  15
  16 class TcpOutChannel(Channel):
  17     '''
  18     Channel bound to a TCP server and listen for data there.
  19     '''
  20     def __init__(self, id4, hostname, port):
  21         Channel.__init__(self)
  22         self.id4 = id4
  23         self.source = get_source_by_id4(id4)
  24         self.hostname = hostname
  25         self.port = port
  26         self.name = '%s:%s' % (self.hostname, self.port)
  27         address_out = (self.hostname, self.port)
  28
  29         # TODO try all IP addresses
  30         # see http://docs.python.org/library/socket.html#example
  31         try:
  32             self.sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
  33             self.sock.connect(address_out)
  34             logging.info("Connected to %s using tcp6", address_out)
  35         except socket.gaierror as err:
  36             logging.error("Can't connet to %s: %s using tcp6", address_out, err)
  37             self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  38             try:
  39                 self.sock.connect(address_out)
  40             except socket.error as err:
  41                 logging.error("Can't connet to %s: %s using tcp4", address_out, err)
  42                 raise
  43             logging.info("Connected to %s using tcp4", address_out)
  44
  45     def fill_buffer(self):
  46         '''
  47         Grab more bytes into self.data internal buffer.
  48         '''
  49         data = self.sock.recv(DEFAULT_MTU)
  50         if not data:
  51             # FIXME: do reconnect
  52             assert data, 'Connection closed'
  53         stats = self.source.stats
  54         stats.npackets += 1
  55         stats.nbytes += len(data)
  56         stats.nbytes_ethernet += len(data) + TCP_HEADER_SIZE
  57         logging.debug('IN %s %s', self.name, repr(data))
  58         self.data += data
  59
  60
  61 class TcpOutServiceIn(ServiceIn):
  62     '''
  63     Service that connects to a TCP server and listen for data there.
  64     '''
  65     def __init__(self, id4, hostname, port):
  66         #ServiceIn.__init__(self, stdout)
  67         self.channel = TcpOutChannel(id4, hostname, port)
  68
  69     def __repr__(self):
  70         return 'TcpOutServiceIn<%s,%s>' \
  71             % (self.channel.hostname, self.channel.port)
  72
  73     def fileno(self):
  74         '''
  75         Returns file descriptor of underlying socket, for os.select().
  76         '''
  77         return self.channel.sock.fileno()
  78
  79     def get_activity(self):
  80         '''
  81         Pool the channel and returns information when ready.
  82         '''
  83         channel = self.channel
  84         channel.fill_buffer()
  85         return channel.name, channel