def socketpair():
     s1, s2 = eunuchs.socketpair.socketpair()
     p, c = (socket.fromfd(s1, socket.AF_UNIX, socket.SOCK_STREAM),
             socket.fromfd(s2, socket.AF_UNIX, socket.SOCK_STREAM))
     os.close(s1)
     os.close(s2)
     return p, c

def fromfd(fd, keep_fd=True):
    """Create a socket from a file descriptor

    socket domain (family), type and protocol are auto-detected. By default
    the socket uses a dup()ed fd. The original fd can be closed.

    The parameter `keep_fd` influences fd duplication. Under Python 2 the
    fd is still duplicated but the input fd is closed. Under Python 3 and
    with `keep_fd=True`, the new socket object uses the same fd.

    :param fd: socket fd
    :type fd: int
    :param keep_fd: keep input fd
    :type keep_fd: bool
    :return: socket.socket instance
    :raises OSError: for invalid socket fd
    """
    family = _raw_getsockopt(fd, socket.SOL_SOCKET, SO_DOMAIN)
    typ = _raw_getsockopt(fd, socket.SOL_SOCKET, SO_TYPE)
    proto = _raw_getsockopt(fd, socket.SOL_SOCKET, SO_PROTOCOL)
    if sys.version_info.major == 2:
        # Python 2 has no fileno argument and always duplicates the fd
        sockobj = socket.fromfd(fd, family, typ, proto)
        sock = socket.socket(None, None, None, _sock=sockobj)
        if not keep_fd:
            os.close(fd)
        return sock
    else:
        if keep_fd:
            return socket.fromfd(fd, family, typ, proto)
        else:
            return socket.socket(family, typ, proto, fileno=fd)

    def _get_systemd_socket(self, address):
        fds = sd.listen_fds()
        if not fds:
            return address
        elif len(fds) > 1:
            raise ValueError('Too many listening sockets', fds)

        if isinstance(address, tuple):
            port = address[1]
            # systemd uses IPv6
            if not sd.is_socket_inet(fds[0], family=socket.AF_INET6,
                                     type=socket.SOCK_STREAM,
                                     listening=True, port=port):
                raise ValueError("FD {} is not TCP IPv6 socket on port {}",
                                 fds[0], port)
            logger.info('Using systemd socket activation on port %i', port)
            sock = socket.fromfd(fds[0], socket.AF_INET6, socket.SOCK_STREAM)
        else:
            if not sd.is_socket_unix(fds[0], socket.SOCK_STREAM,
                                     listening=True, path=address):
                raise ValueError("FD {} is not Unix stream socket on path {}",
                                 fds[0], address)
            logger.info('Using systemd socket activation on path %s', address)
            sock = socket.fromfd(fds[0], socket.AF_UNIX, socket.SOCK_STREAM)

        if sys.version_info[0] < 3:
            # Python 2.7's socket.fromfd() returns _socket.socket
            sock = socket.socket(_sock=sock)
        return sock

 def __init__(self, thread_index):
     protocol_str = os.getenv('CLOUDI_API_INIT_PROTOCOL')
     if protocol_str is None:
         raise invalid_input_exception()
     buffer_size_str = os.getenv('CLOUDI_API_INIT_BUFFER_SIZE')
     if buffer_size_str is None:
         raise invalid_input_exception()
     if protocol_str == 'tcp':
         self.__s = socket.fromfd(
             thread_index + 3, socket.AF_INET, socket.SOCK_STREAM
         )
         self.__use_header = True
     elif protocol_str == 'udp':
         self.__s = socket.fromfd(
             thread_index + 3, socket.AF_INET, socket.SOCK_DGRAM
         )
         self.__use_header = False
     elif protocol_str == 'local':
         self.__s = socket.fromfd(
             thread_index + 3, socket.AF_UNIX, socket.SOCK_STREAM
         )
         self.__use_header = True
     else:
         raise invalid_input_exception()
     self.__initializtion_complete = False
     self.__size = int(buffer_size_str)
     self.__callbacks = {}
     self.__send(term_to_binary(OtpErlangAtom('init')))
     (self.__prefix,
      self.__timeout_async, self.__timeout_sync,
      self.__priority_default,
      self.__request_timeout_adjustment) = self.__poll_request(False)

 def closesocketfunction(curlfd):
     called['called'] = True
     # Unix only
     #os.close(curlfd)
     # Unix & Windows
     socket.fromfd(curlfd, socket.AF_INET, socket.SOCK_STREAM).close()
     return 0

 def accept(self):
     logging.debug('connection accepted')
     conn = self.listener.nextPendingConnection()
     if hasattr(socket, 'fromfd'):
         socket.fromfd(conn.socketDescriptor(), socket.AF_INET, socket.SOCK_STREAM).setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 2**22)
     self.conns.append(conn)
     conn.readyRead.connect(lambda: self.read_from(conn))

def hook(uds_datap):
  outer_sock = socket.fromfd(uds_datap.contents.outer_sock, socket.AF_INET,
                              socket.SOCK_STREAM, 0)
  inner_sock = socket.fromfd(uds_datap.contents.inner_sock, socket.AF_INET,
                              socket.SOCK_STREAM, 0)
  custom_hook(outer_sock, inner_sock)
  hookffi.teardown(uds_datap)
  return 0

    def test_fromfd(self):
        msg = 'hello world'
        x, y = socket.socketpair()
        xx = socket.fromfd(x.fileno(), x.family, socket.SOCK_STREAM)
        x.close()
        yy = socket.fromfd(y.fileno(), y.family, socket.SOCK_STREAM)
        y.close()

        xx.sendall(msg)
        xx.close()
        read = yy.makefile().read()
        self.assertEqual(msg, read)

def make_server(handler, host='localhost', port=8990):
    pool = Pool(100)
    server = StreamServer((host, port), handler, spawn=pool)

    # graceful startup
    listener_fd, worker_fds = umgmt.graceful_startup(server, 'account', accepted)
    if listener_fd is not None:
        server.set_listener(socket.fromfd(listener_fd, socket.AF_INET, socket.SOCK_STREAM))
    if worker_fds:
        for w in worker_fds:
            s = socket.fromfd(w, socket.AF_INET, socket.SOCK_STREAM)
            gevent.spawn(handler, s, None)

    return server

    def NewConnection(self, path, fd, properties):
        self.fd = fd.take()
        print("NewConnection(%s, %d)" % (path, self.fd))


        server_sock = socket.fromfd(self.fd, socket.AF_UNIX, socket.SOCK_STREAM)
        server_sock.setblocking(1)
        server_sock.send("This is Edison SPP loopback test\nAll data will be loopback\nPlease start:\n")

        try:
            while True:
                print("start recv\n")
                data = server_sock.recv(1024)
                print("received: %s" % data)
                fsensor = '/tmp/sensors'
                if os.path.isfile(fsensor):
                    print("1\n")
                    f = open(fsensor, 'r')
                    print("2\n")
                    rsp = f.readline()
                    print("3\n")
                    f.close()
                    print("rsp: %s\n" % rsp)
                    server_sock.send("%s\n" % rsp)
                    print("4\n")
                else:
                    print("sensor file not exists\n")
                    server_sock.send("z\n")
                    print("5\n")
        except IOError:
            print ("IOError")
            pass

        server_sock.close()
        print("all done")

    def _init(self):
        self._pids.clear()

        # getting the initial list of watchers/pids
        res = self.client.send_message('list')

        for watcher in res['watchers']:
            if watcher in ('circusd', 'circushttpd', 'circusd-stats'):
                # this is dealt by the special 'circus' collector
                continue

            pid_list = self.client.send_message('list', name=watcher)
            pids = pid_list.get('pids', [])
            for pid in pids:
                self._append_pid(watcher, pid)

        # getting the circus pids
        self.circus_pids = self.get_circus_pids()
        if 'circus' not in self._callbacks:
            self._add_callback('circus')
        else:
            self._callbacks['circus'].start()

        # getting the initial list of sockets
        res = self.client.send_message('listsockets')
        for sock in res.get('sockets', []):
            fd = sock['fd']
            address = '%s:%s' % (sock['host'], sock['port'])
            # XXX type / family ?
            sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)
            self.sockets.append((sock, address, fd))

        self._add_callback('sockets', kind='socket')

    def in_q_cb(self, watcher, revents):

        try:
            val = self.in_q.get()
            #val = self.in_q.get(True,interval)
            logging.debug("ServerWorker[{0}:{1}]: Received inQ event!".format(os.getpid(),self.name))
            if type(val) == type((1,)):

                # Construct a proper socket object from the socket FD
                client_socket_handle,client_address = val
                client_fd = rebuild_handle(client_socket_handle)
                client_socket = socket.fromfd(client_fd, socket.AF_INET, socket.SOCK_STREAM)

                logging.debug("ServerWorker[{0}:{1}]: Adding connection [{2}] from [{3}].".format(os.getpid(),self.name,self.client_count,client_address))

                self.client_count += 1
                self.cnxns[client_address] = Connection(client_socket, client_address, self.loop, self.client_count, self)

                self.reset(pyev.EV_READ)

            elif type(val) == type("") and val == "quit":

                logging.info("ServerWorker[{0}:{1}]: Received quit message!".format(os.getpid(),self.name))
                self.stop()

        except Queue.Empty:
            # Timed-out, carry on
            pass

 def __init__(self):
     self._send, self._recv = _UNPATCHED_SOCKETPAIR()
     # Assume monkey patching if socket.socketpair is different.
     self._patched = socket.socketpair != _UNPATCHED_SOCKETPAIR
     if self._patched:
         self._send_patched_lock = thread.allocate_lock()
         self._send_patched = socket.fromfd(self._send.fileno(),
             self._send.family, self._send.type)
         self._send_patched.settimeout(None)
         self._recv_patched_lock = thread.allocate_lock()
         self._recv_patched = socket.fromfd(self._recv.fileno(),
             self._recv.family, self._recv.type)
     self._send.settimeout(None)
     self._recv_lock = _UNPATCHED_ALLOCATE_LOCK()
     self._items = []
     self._write_byte = True

    def __init__(self, plugin, args):
        self.cmdline = False
        if len(args) == 2 and PluginRunner._is_number(args[1]):
            try:
                fd = int(args[1])
                self.tp = _transport.TransPort(
                    socket.fromfd(fd, socket.AF_UNIX, socket.SOCK_STREAM))

                #At this point we can return errors to the client, so we can
                #inform the client if the plug-in fails to create itself
                try:
                    self.plugin = plugin()
                except Exception as e:
                    exception_info = sys.exc_info()

                    self.tp.send_error(0, -32099,
                                       'Error instantiating plug-in ' + str(e))
                    raise exception_info[1], None, exception_info[2]

            except Exception:
                error(traceback.format_exc())
                error('Plug-in exiting.')
                sys.exit(2)

        else:
            self.cmdline = True
            cmd_line_wrapper(plugin)

    def get_request_socket(self, env):
        if not self.ca:
            return None

        sock = None

        if env.get('uwsgi.version'):  # pragma: no cover
            try:
                import uwsgi
                fd = uwsgi.connection_fd()
                conn = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)
                try:
                    sock = socket.socket(_sock=conn)
                except:
                    sock = conn
            except Exception as e:
                pass
        elif env.get('gunicorn.socket'):  # pragma: no cover
            sock = env['gunicorn.socket']

        if not sock:
            # attempt to find socket from wsgi.input
            input_ = env.get('wsgi.input')
            if input_:
                if hasattr(input_, '_sock'):  # pragma: no cover
                    raw = input_._sock
                    sock = socket.socket(_sock=raw)  # pragma: no cover
                elif hasattr(input_, 'raw'):
                    sock = input_.raw._sock

        return sock

    def _setupSocket(self):
        if self._bindAddress is None:  # Run as a normal FastCGI?

            sock = socket.fromfd(FCGI_LISTENSOCK_FILENO, socket.AF_INET,
                                 socket.SOCK_STREAM)
            try:
                sock.getpeername()
            except OSError as e:
                if e.errno != errno.ENOTCONN:
                    raise
        else:
            # Run as a server
            oldUmask = None
            if isinstance(self._bindAddress, str):
                # Unix socket
                sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
                try:
                    os.unlink(self._bindAddress)
                except OSError:
                    pass
                if self._umask is not None:
                    oldUmask = os.umask(self._umask)
            else:
                # INET socket
                assert type(self._bindAddress) is tuple
                assert len(self._bindAddress) == 2
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            close_on_exec(sock.fileno())
            sock.bind(self._bindAddress)
            sock.listen(socket.SOMAXCONN)

            if oldUmask is not None:
                os.umask(oldUmask)
        return sock

	def run(self):
		url = "/%s" % '/'.join([self.chan, self._session, self._id, "xhr_streaming"])
		if DEBUG:
			print "TRYING URL %s" % url

		if self.use_ssl:
			con = httplib.HTTPSConnection(self.host, self.port)
		else:
			con = httplib.HTTPConnection(self.host, self.port)
		
		con.request('POST', url)

		r = con.getresponse()
		self.sock = socket.fromfd(r.fileno(), socket.AF_INET, socket.SOCK_STREAM)
		
		data = 1
		while data:
			data = self.sock.recv(1)
			if data == 'o':
				#if DEBUG: print "CONNECTION!"
				pass

			if data == 'c':
				#if DEBUG: print "DISCONNECTION"
				pass

			if data in ('m', 'a'):
				msg = self.sock.recv(1000)
				self.route_annex_channel_message(msg)

		sleep(0)
		if DEBUG: print "Channel to task %s closed." % self.chan

def is_socket(fd):
    """ Determine whether the file descriptor is a socket.

        :param fd: The file descriptor to interrogate.
        :return: ``True`` iff the file descriptor is a socket; otherwise
            ``False``.

        Query the socket type of `fd`. If there is no error, the file is a
        socket.

        """
    result = False

    file_socket = socket.fromfd(fd, socket.AF_INET, socket.SOCK_RAW)

    try:
        socket_type = file_socket.getsockopt(
                socket.SOL_SOCKET, socket.SO_TYPE)
    except socket.error as exc:
        exc_errno = exc.args[0]
        if exc_errno == errno.ENOTSOCK:
            # Socket operation on non-socket.
            pass
        else:
            # Some other socket error.
            result = True
    else:
        # No error getting socket type.
        result = True

    return result

    def fileDescriptorsReceived(self, fds, message):
        if len(fds) == 1:
            fd = fds[0]

            # Note that we hardcode IPv4 here!
            sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)

            # PROBE: received fd; see porter.py
            self.debug("[fd %5d] (ts %f) received fd from %d, created socket",
                       sock.fileno(), time.time(), fd)

            # Undocumentedly (other than a comment in
            # Python/Modules/socketmodule.c), socket.fromfd() calls dup() on
            # the passed FD before it actually wraps it in a socket object.
            # So, we need to close the FD that we originally had...
            os.close(fd)

            try:
                peeraddr = sock.getpeername()
            except socket.error:
                self.info("Socket disconnected before being passed to client")
                sock.close()
                return

            # Based on bits in tcp.Port.doRead()
            addr = address._ServerFactoryIPv4Address('TCP',
                peeraddr[0], peeraddr[1])
            protocol = self.childFactory.buildProtocol(addr)

            self._connectionClass(sock, protocol, peeraddr, message)
        else:
            self.warning("Unexpected: FD-passing message with len(fds) != 1")

	def NewConnection(self, path, fd, properties):
		self.fd = fd.take()
		print("NewConnection(%s, %d)" % (path, self.fd))


		server_sock = socket.fromfd(self.fd, socket.AF_UNIX, socket.SOCK_STREAM)
		server_sock.setblocking(1)
		server_sock.send("This is Edison SPP loopback test\nAll data will be loopback\nPlease start:\n")

		try:
		    while True:
		        data = server_sock.recv(1024)
		        print("received: %s" % data)

			server_sock.send("looping back: %s\n" % data)
			server_sock.send("OK\n")
			
		        file_temp = open('example.txt', 'w')
			file_temp.write(data)
			file_temp.close()
		except IOError:
		    pass

		server_sock.close()
		print("all done")