def main():
    setproctitle('graph4py.rpc')
    parser = argparse.ArgumentParser(description='Run Graph4Py graph server')
    parser.add_argument(
        '--version',
        '-v',
        action='version',
        version=__version__
    )
    parser.add_argument('host')
    parser.add_argument('port', type=int)
    parser.add_argument('backend')
    parser.add_argument('path')
    parser.add_argument('--authkey', '-k', action='store')
    parser.add_argument('--graph', '-g', action='store')
    args = parser.parse_args()

    graph_class = import_object(args.graph) if args.graph else None

    database = Graph4Py(args.backend, args.path, args.authkey, graph_class)

    def signal_handler(s, frame):
        database.close()
        sys.exit(0)
    signal.signal(signal.SIGINT, signal_handler)

    listener = Listener((args.host, args.port), family='AF_INET')
    print 'Running on %s:%s with %s' % (args.host, args.port, database.graph)

    while True:
        connection = listener.accept()
        database.process(connection)

class CountApiServer(threading.Thread):

    def __init__(self, port, parent):
        super(CountApiServer, self).__init__()

        self.parent = parent
        self.__listener = Listener(('127.0.0.1', port))

    def run(self):
        conn = self.__listener.accept()
        while 1:
            if conn.closed:
                conn = self.__listener.accept()
            try:
                msg = conn.recv()
            except EOFError:
                conn.close()
                continue
            else:
                if msg == 'quit':
                    conn.close()
                    break
            conn.send(self.parent.key_count)
        self.__listener.close()
        self.parent.stop()

    def rpc(self, dest, method, args=None, kwargs=None):
        """ Make an RPC call to the named subscriber, expecting a
        response.  This opens a
        :class:`multiprocessing.connection.Listener` and passes the
        Listener address to the child as part of the RPC call, so that
        the child can connect to the Listener to submit its results.

        Listeners are reused when possible to minimize overhead.
        """
        try:
            listener = self._available_listeners.get_nowait()
            self.logger.debug("Reusing existing RPC listener at %s" %
                              listener.address)
        except Empty:
            listener = Listener()
            self.logger.debug("Created new RPC listener at %s" %
                              listener.address)
        self._blocking_listeners.append(listener)
        try:
            self._queues[dest].put((listener.address,
                                    (method, args or [], kwargs or dict())))
            conn = listener.accept()
            self._blocking_listeners.remove(listener)
            try:
                while not self._terminate.is_set():
                    if conn.poll(self.poll_wait):
                        return conn.recv()
            finally:
                conn.close()
        finally:
            self._available_listeners.put(listener)

 def loop(self):
     listener = Listener(self.address, backlog=5, authkey=self.authkey)
     logging.warn('DB server started with %s, listening on %s:%d...',
                  sys.executable, *self.address)
     self.thread.start()
     cmd = None
     try:
         while cmd != 'stop':
             try:
                 conn = listener.accept()
             except KeyboardInterrupt:
                 break
             except:
                 # unauthenticated connection, for instance by a port
                 # scanner such as the one in manage.py
                 continue
             cmd_ = conn.recv()  # a tuple (name, arg1, ... argN)
             cmd, args = cmd_[0], cmd_[1:]
             if cmd.startswith('@'):  # slow command, run in process
                 cmd = cmd[1:]  # strip @
                 proc = Process(
                     target=run_command, name=cmd, args=(cmd, args, conn))
                 proc.start()
                 logging.warn('Started %s%s in process %d',
                              cmd, args, proc.pid)
             else:
                 queue.put((conn, cmd, args))
     finally:
         listener.close()
         self.thread.join()

    def _serve(self, port, path):
        data = []
#        print 'Server', os.getpid()
        address = ('localhost', port)
        listener = Listener(address)
        loop = True
        while loop:
            conn = listener.accept()
            command, payload = conn.recv()
            response_payload = None
            if command == 'OPEN':
                #print command, payload
                path = payload
            elif command == 'WRITE':
                #print command, payload
                data.append(payload)
            elif command == 'FETCH':
                #print command, payload
                response_payload = data
            elif command == 'FLUSH':
                #print command
                json.dump(data, open(path, 'w'))
            elif command == 'CLOSE':
                #print command
                #json.dump(data, open(path, 'w'))
                loop = False
            conn.send(('OK', response_payload))
            conn.close()
        listener.close()

def main():
    _setproctitle('structurarium.graph')
    parser = argparse.ArgumentParser(description='Run Structurarium graph server')
    parser.add_argument('--version', '-v', action='version', version=__version__)
    parser.add_argument('host')
    parser.add_argument('port', type=int)
    parser.add_argument('path')
    parser.add_argument('--authkey', '-k', action='store')
    parser.add_argument(
        '--worker',
        '-w',
        action='store',
        type=int,
        help='default is set to the number of CPU'
    )
    args = parser.parse_args()
    listener = Listener((args.host, args.port), family='AF_INET')
    database = Graph(args.path, authkey=args.authkey)
    pool = Pool(processes=args.worker)

    print 'Running on %s:%s' % (args.host, args.port)
    while True:
        connection = listener.accept()
        connection = reduce_connection(connection)
        # pool.apply_async(connect, [database, connection])
        connect(database, connection)

def worker(address, port, password):
    listener = Listener(address, authkey=password)
    conn = listener.accept()
    print 'connection accepted from', listener.last_accepted
    conn.send_bytes('hello')
    conn.close()
    listener.close()

def objectLogger(params):
    """ Send data (dict, str, ...) into a storage """
    mode = params['mode']
    maxlen = params['maxlen']
    file_format = params['format']
    home_dir = params['storage_dir']
    server = Listener((params['ip'], int(params['port'])))
    storage = PersistentDeque(
        home_dir=home_dir,
        maxlen=maxlen,
        file_format=file_format,
        mode=mode,
    )
    while True:
        conn = server.accept()
        while True:
            try:
                data = conn.recv()
            except EOFError:
                break
            if data:
                storage.append(data)
            else:
                storage.sync(id_generator())
                storage = PersistentDeque(
                    home_dir=home_dir,
                    maxlen=maxlen,
                    file_format=file_format,
                    mode=mode,
                )
        conn.close()

def rpc_server(handler, address, authkey):
    sock = Listener(address, authkey=authkey)
    while True:
        client = sock.accept()
        t = Thread(target=handler.handle_connection, args=(client,))
        t.daemon = True
        t.start()

    def run(self):
        listener = Listener(self.address, self.family)

        while True:
            conn = listener.accept()
            t = threading.Thread(target=self._handle_conn, args=(conn,))
            t.daemon = True
            t.start()

 def close(self, blockUntilClosed = True):
     '''close may wait 1 second.'''
     Listener.close(self)
     if self.establish_connection_to_close_lock.acquire(False):
         if not blockUntilClosed:
             thread_start_new(self._connect_to_pipe_to_close_it, ())
         else:
             self._connect_to_pipe_to_close_it()

def send(msg):
    address = ('localhost', 6000)
    listener = Listener(address)
    conn = listener.accept()
    conn.send(msg)

    conn.close()
    listener.close()

 def run(self):
     serv = Listener(self.address,authkey=self.authkey)
     while True:
         try:
             client = serv.accept()
             DispatchClientTask(client).start()
         except Exception as e:
             self.log.info("Error : %s", e, exc_info=True)

 def start(self):
     address = ('localhost', 6000)
     listener = Listener(address)
     conn = listener.accept()
     while True:
         msg = conn.recv()
         self.log_id += 1
         self.log_file.add_row(LogEntry([self.node_id, self.log_id, msg]))

 def _listenThread(self):
     address = ('localhost', self.port)     # family is deduced to be 'AF_INET'
     listener = Listener(address, authkey=self.pwd)
     while True:
         conn = listener.accept()
         self._printDebug('MultiBus,Listner: connection accepted from' + str(listener.last_accepted))
         p = Process(target=self._connThread, args=(conn,))
         p.start()

def echo_server(address, authkey):
    serv = Listener(address, authkey=authkey)
    while True:
        try:
            client = serv.accept()
            echo_client(client)
        except Exception:
            traceback.print_exc()

class SingleInstanceApp(BaseJobManager):
    def __init__(self, name, session=''):
        BaseJobManager.__init__(self)
        if sys.platform == 'win32':
            self.address = r'\\.\pipe\single_instance_app.'
        else:
             self.address = '/var/tmp/single_instance_app.pipe'
        trans = string.maketrans('\\ ', '__')
        self.address += '.'+name.translate(trans)
        if session and len(session)>0:
            self.address += session.translate(trans)

        self.accepting_connection = False


    def _accept_connection(self):
        while self.accepting_connection:
            conn = self.listener.accept()
            args = conn.recv()
            conn.close()
            self.queue_launch_instance(*args)

    def try_start_instance(self, *args):
        try:
            conn = Client(self.address, authkey='secret password')
            conn.send(args)
            conn.close()
            #print "sent my args(%s)"%(', '.join(args))
            return False
        except:
            self.queue_launch_instance(*args)

            self.listener = Listener(self.address, authkey='secret password')

            self.accepting_connection = True

            self.listener_tread = threading.Thread(target=self._accept_connection)
            self.listener_tread.daemon = True #because of listener.accept() blocks
            self.listener_tread.start()
            return True

    def stop_accepting(self):
        self.accepting_connection = False
        self.listener.close()
        self.listener_tread.join(0.01)

    def queue_launch_instance(self, *args):
        fancy_log(sys.stdout, 'queued instance launch with args: '+str(args))
        self.add_job(*args)

    #override
    #def do_the_job(self, *args):
    #    print 'calls instance with args: ',args
    #    self.update_job()

    def job_is_updated(self, *args, **kwargs):
        fancy_log(sys.stdout,'calls instance with args: '+str(args)+'\n kwargs: '+str(kwargs))
        pass

def client_listener():
    cl = Listener(address=local_listener[0], authkey=local_listener[1])
    print '.............client listener starting' 
    print '.............accepting conexions'
    while True:
        conn = cl.accept()
        print '.............connection accepted from', cl.last_accepted        
        m = conn.recv()
        print '.............message received from server', m 

 def run(self):
     listen = Listener(self._address)
     print 'simplepycache server start to listen at', self._address
     cleaner = self.Cleaner(self)
     cleaner.deamon = True
     cleaner.start()        
     while True:
         conn = listen.accept()
         self.Worker(self, conn).start()

 def run(self):
     server = Listener( ('', 25000) )
     logging.getLogger("thread-listener").info("Connection listener initialized.")
     while self.RUN:
         try:
             client = server.accept()
             logging.getLogger("thread-listener").debug("Client connecting...")
             self.echo_client(client)
         except Exception:
             traceback.print_exc()