gSocket.py

Go to the documentation of this file.

#!/usr/local/bin/python
#
#                               Copyright 2004
#                                     by
#                        The Board of Trustees of the
#                     Leland Stanford Junior University.
#                            All rights reserved.
#


__facility__ = "Online"
__abstract__ = "Socket class"
__author__   = "Amedeo Perazzo <perazzo@slac.stanford.edu> SLAC - GLAST I&T/Online"
__date__     = ("$Date: 2005/09/22 00:29:31 $").split(' ')[1]
__version__  = "$Revision: 1.2 $"
__release__  = "$Name: HEAD $"
__credits__  = "SLAC"

import LICOS.copyright_SLAC

import socket
import threading

class ConnectState:
  Disconnected = 0
  Waiting      = 1
  Connected    = 2

class Socket(object):
  TIMEOUT = 5

  def __init__(self, portnumber):
    self.__port = portnumber
    self.__sock = None
    self.__host = None
    self.__lock = threading.Lock()

  def connect(self, host=None, connectors=1):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(self.TIMEOUT)  # TMOs are not handled correctly
    try:
      if host is None:
        # listening socket
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(self.TIMEOUT)  # TMOs are not handled correctly
        sock.bind(('', port))
        sock.listen(connectors)
        self.__sock,addr = sock.accept()
      else:
        # external socket
        self.__sock = sock
        self.__sock.connect((host, self.__port))
        self.__host = host
          
    except socket.error, msg:
      self.disconnect()
      raise RuntimeError, \
            'gSocket: unable to connect to %s on port %d: %s' \
            %(host, self.__port, str(msg))
    ss = 2*1024*1024
    self.__sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, ss)

  def disconnect(self):
    if self.isConnected():
      self.__sock.close()
      self.__sock = None

  def isConnected(self):
    return (self.__sock is not None)

  def socket(self):
    return self.__sock

  def setSocket(self, sock):
    self.__sock = sock

  def port(self):
    return self.__port

  def setPort(self, port):
    self.__port = port

  def lock(self):
    self.__lock.acquire()

  def release(self):
    self.__lock.release()

  def gettimeout(self):
    return self.__sock.gettimeout()

  def settimeout(self, seconds):
    self.__sock.settimeout(seconds)

  def send(self, packet):
    return self.__sock.send(packet)

  def recv(self, size):
    return self.__sock.recv(size)

  def sendall(self, packet):
    return self.__sock.sendall(packet)

  def recvall(self, size):
    # The following is not supported on Windows, at least.
    # return self.__sock.recv(size, socket.MSG_WAITALL)

    data = ''
    remaining = size
    sock = self.__sock
    while remaining > 0:
      received = sock.recv(remaining)
      reclen = len(received)
      if reclen == 0: return received
      remaining -= reclen
      data += received
    return data

  def purge(self):
    """!\brief Purge all stale data out of the socket.
    """
    max  = 64*1024
    sock = self.__sock
    tmo  = sock.gettimeout()
    sock.settimeout(0.1)
    received = ''
    size = 0
    while size != 0:
      try:
        received = sock.recv(max)
      except socket.error, e:
        if str(e) != "timed out":  raise # Throw away "timed out" exceptions
      size = len(received)

    sock.settimeout(tmo)
    
class ReconnectingSocket(Socket):
  def __init__(self, port, host):
    Socket.__init__(self, port)
    self.__name = name
    self.__host = host
    
  def send(self, packet):
    self.lock()
    rv = super(ReconnectingSocket, self).send(packet)
    self.release()
    return rv

  def recv(self, size):
    self.lock()
    rv = super(ReconnectingSocket, self).recv(size)
    self.release()
    return rv

  def sendall(self, packet):
    self.lock()
    rv = super(ReconnectingSocket, self).sendall(packet)
    self.release()
    return rv

  def recvall(self, packet):
    self.lock()
    rv = super(ReconnectingSocket, self).recvall(packet)
    self.release()
    return rv

  def connect(self, host=None):
    """!\brief connect

    Connect to the socket
    """
    self.__connection_state = ConnectState.Waiting
    self.__tries = 0
    self.__host = host
    self.__connect()
    
  def disconnect(self):
    """!\brief disconnected

    """
    self.__disconnect()

  def timedout(self):
    """!\brief timedout
    """
    self.__disconnect()

  def __disconnect(self):
    # disconnect from the server
    super(ReconnectingSocket, self).disconnect()
    # deal with the outcome of disconnecting
    self.__connection_notthere()
    
  def __connection_notthere(self):
    # Do this once...
    if self.__connection_state == ConnectState.Connected:
      log.error("Socket disconnected, trying to reconnect" )
      self.__connection_state = ConnectState.Waiting
      self.__connect()
    
  def __connect(self):
    # spawn the connecting thread in the background
    connecthdlr = threading.Thread(None, self.__connection_handler, 'ConnectHandler',())
    connecthdlr.setDaemon(True)
    connecthdlr.start()

  def __connection_handler(self):
    while (True):
      try:
        super(ReconnectingSocket, self).connect(self.__host)
        log.info("Socket is now connected")
        self.__connection_state = ConnectState.Connected
        self.__tries = 0
        break
      except RuntimeError, msg:
        if self.__tries == 0:
          log.error(msg)
        self.__tries += 1
        state = self.__connection_state
        if state == ConnectState.Disconnected: break
        time.sleep(3)
    
    
# EOF

---