ppcg_sockets.py

#! /usr/bin/env python
#-*- coding: utf-8 -*-
"""
 Source: http://forum.python.org.pl/index.php?topic=1607.msg7913#msg7913
"""

import socket
import sys
import os
import datetime
import select

import Queue

# Albo jedno, albo drugie - łączenie wątków i procesów to 
# proszenie się o kłopoty
import threading
import multiprocessing

# Do logów wygodnie użyć modułu logging, który jest bezpieczny ze względu na wątki
import logging
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
logger = logging.getLogger()

def exc_handler(exc):
    """Tlumaczenie wyjatkow"""
    e={98:"Adres już jest używany"}

    if e.has_key(exc[0]):
        return e[exc[0]]
    else:
        return exc

class Server():
    """Server - zarządzanie połączeniami"""

    S_QUIT = 0
    S_INIT = 1
    S_RUN = 2

    def __init__(self):
        """Konstruktor"""
        self.state = Server.S_INIT  # Pole do przechowywania stanu
        self.app_queue = multiprocessing.Queue()  # FIFO aplikacji
        self.app_queue_fd = self.app_queue._reader  # Obiekt app_queue dla selecta
        self.con_queue = Queue.Queue()  # FIFO połączeń
        self.threads = []  # Kontener na wątki robocze
        self.host = ''  # Nazwa hosta
        self.port = 9765  # Port do nasluchiwania
        self.backlog = 5  # Max ilosc polaczen
        self.family = socket.AF_INET
        self.typ = socket.SOCK_STREAM
        self.socket = None  # Kontener na gniazdo
    
        # Listy obiektw dla selecta
        self.rlist = [sys.stdin, self.app_queue_fd]
        self.wlist = []
        self.xlist = []
        self.select_timeout = 0

    def open_socket(self):
        """Tworzenie socket'u"""
        try:
            self.socket = None
            self.socket = socket.socket(self.family, self.typ)
            logger.info("Przygotowuje gniazdo sieciowe")
      
            self.socket.bind((self.host, self.port))            
            logger.info("Port: %d został zajęty", self.port)
            logger.info("Zaczynam nasłuchiwać(max %d połączeń)", self.backlog)
            self.socket.listen(self.backlog)                    
        except socket.error, exc:
            logger.exception("Nie udało się ustawić gniazda serwera")
            if self.socket:
                self.socket.close()
            self.socket = None

    def run(self):
        """Główna pętla servera"""
        self.open_socket()  # Otwarcie socketa
        self.rlist.append(self.socket)  # Dodanie socketa do listy odczytu
        
        logger.info("Tworzenie wątków roboczych")
        for i in range(3):
            n = "X%02d" % i
            t = Handler(self.con_queue, self.app_queue, n)
            t.start()
            self.threads.append(t)
    
        ## odpalanie głównej pętli programu
        logger.info("Czekam na połączenia")
        logger.info("....................")
        self.state = Server.S_RUN
        
        while(self.state != Server.S_QUIT):
            rwait, wwait, xwait = select.select(
                                self.rlist,
                                self.wlist,
                                self.xlist,
                                self.select_timeout)

            for s in rwait:
                if s == self.socket:
                    self.con_queue.put(self.socket.accept())
                
                if s == sys.stdin:
                    line = s.readline()
                    line = line.strip()
                    line = line.split()
                    if line:
                        fun_name = "self."+line[0]+"_stdin"
                        fun_args = line[1:]
                        try:
                            _ = callable(eval(fun_name))
                        except AttributeError:
                            fun_name = "self.NotImpl"
                        
                        # odpalanie funkcji przypisanej do komendy
                        eval(fun_name)(fun_args)
        
                if s == self.app_queue_fd:
                    app_data = self.app_queue.get()
                    if app_data.__class__ == tuple().__class__:
                        fun_name = "self."+app_data[0]
                        fun_c_id = app_data[1]
                        fun_args = app_data[2]
                    try:
                        _ = callable(eval(fun_name))
                    except AttributeError:
                        # TODO: kontener na błędne ramki
                        fun_name = "self.XS00002"  # ramka servera NotImpl
                        # odpalanie funkcji przypisanej do ramki
                    eval(fun_name)(fun_c_id, fun_args)
        
        logger.info("Zamykam socket")
        
        # zamykanie wątków
        for t in self.threads:
            t.exit()
            t.join()
        self.socket.close()

    """
      Funkcje przypisane do komend podawanych z stdin
      W __doc__ wiadomości dla funkcji help:
      do pierwszego @@@ krótki opis,
      następnie opis przy wywołaniu helpa dla tej funkji
      każde kolejne jest dla argumentów w postaci:
      @@@nazwa_arg@@@help@@@nazwa2@@@help2
    """
    
    def NotImpl(self, args):
        """Funkcja jeszcze nie zaimplementowana"""
        logger.info("Polecenie nieznane lub jeszcze nie zaimplementowane")

    def quit_stdin(self, args):
        """Zamknięcia servera"""
        self.state = Server.S_QUIT
        
        # Wyłącz wątki potomne
        for t in self.threads:
            t.exit()

    def help_stdin(self, args):
        """Wyświetla dostpępne polecenia, wpisz help polecenie po dokładniejsze informacje"""
        ml = dir(self)
        ml.sort()
        if args:
            if len(args)==1:
                try:
                    _ = ml.index(args[0]+"_stdin")
                    help =  eval("self."+args[0]+"_stdin").__doc__.split("@@@")[1]
                    print "%s -> %s" %(args[0], help)
                except ValueError:
                    print "Nie ma takiego polecenia"
            elif len(args)==2:
                try:
                    _ = ml.index(args[0]+"_stdin")
                    help = eval("self."+args[0]+"_stdin").__doc__.split("@@@")[2:]
                    index = help.index(args[1])
                    help = help[index+1]
                    print "%s -> %s" % (args[0]+" "+args[1], help)
                except ValueError:
                    print "Nie ma takiego polecenia"
        else:
            print "Dostępne polecenia to:"
            for m in ml:
                if callable(eval("self."+str(m))) and str(m)[-6:]=="_stdin":
                    help = eval("self."+str(m)).__doc__.split("@@@")[0]
                    print "%s -> %s" % (m[:-6], help)

    """
      Funkcje do obsługi ramek przychodzących
    """
  
    def XS00002(self, c_id, data):
        """ NotImpl """
        log("Ramka nieznane lub jeszcze nie zaimplementowane")

class Handler(threading.Thread):
    """Wątek roboczy servera"""

    S_QUIT = 0
    S_INIT = 1
    S_GET_CONN = 2
    S_RECV = 3
    S_SEND = 4

    def __init__(self, req_queue, app_queue, name):
        # Odpalenie __init__ przodka == threading.Thread.__init__(self)
        super(Handler, self).__init__()
        self.name = name
        self.c_id = "X%02d" %(int(self.name[-1]),)
        self.buf_size = 8192
        self.req_queue = req_queue
        self.app_queue = app_queue
        self.state = Handler.S_INIT
        self.addr = (None, None)
        self.send_buff = None

    def run(self):
        self.state = Handler.S_GET_CONN
        while(self.state != Handler.S_QUIT):
            try:
                client = self.req_queue.get_nowait()
            except Queue.Empty:
                client = None
            
            if client:
                # ustawiamy clienta na nie blokujacego
                client[0].setblocking(False)
                # pobranie adresu
                self.addr = client[1]
                self.state = Handler.S_RECV
                while(self.state == Handler.S_RECV or self.state == Handler.S_SEND):
                    while(self.state == Handler.S_RECV):
                        # odbieranie danych do clienta
                        try:
                            data = client[0].recv(self.buf_size)
                        except socket.error:
                            data = None
                        if not data: continue
                            # tutaj obsługa danych przychodzących - nieistotana dla problemu
                    if self.state == Handler.S_SEND:
                        # wysylanie inforamcji
                        client[0].send(self.send_buff)
                        self.state = Handler.S_RECV
                logging.info("Koniec")
                self.addr = (None, None)
                self.c_id = "X%02d" % int(self.name[-1])

    def exit(self):
        self.state =  Handler.S_QUIT

def main():
    """Glowna petla modulu"""
    app = Server()
    app.run()

## odpalanie modulu
if __name__ == "__main__":
    main()