Python socket.fromshare方法代码示例

本文整理汇总了Python中socket.fromshare方法的典型用法代码示例。如果您正苦于以下问题:Python socket.fromshare方法的具体用法?Python socket.fromshare怎么用?Python socket.fromshare使用的例子?那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在模块socket的用法示例。

在下文中一共展示了socket.fromshare方法的7个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: testTypes

# 需要导入模块: import socket [as 别名]
# 或者: from socket import fromshare [as 别名]
def testTypes(self):
        families = [socket.AF_INET, socket.AF_INET6]
        types = [socket.SOCK_STREAM, socket.SOCK_DGRAM]
        for f in families:
            for t in types:
                try:
                    source = socket.socket(f, t)
                except OSError:
                    continue # This combination is not supported
                try:
                    data = source.share(os.getpid())
                    shared = socket.fromshare(data)
                    try:
                        self.compareSockets(source, shared)
                    finally:
                        shared.close()
                finally:
                    source.close() 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:20,代码来源:test_socket.py


示例2: start_worker

# 需要导入模块: import socket [as 别名]
# 或者: from socket import fromshare [as 别名]
def start_worker(pipeout):
    parse_config()
    httpd = ThreadedTCPServer((
        State.config.get("proxy", "listen").strip(),
        State.config.getint("proxy", "port")), Proxy, bind_and_activate=False)
    mainsock = socket.fromshare(pipeout.recv())
    httpd.socket = mainsock
    print_banner()
    serve_forever(httpd) 
开发者ID:genotrance,项目名称:px,代码行数:13,代码来源:px.py


示例3: run_pool

# 需要导入模块: import socket [as 别名]
# 或者: from socket import fromshare [as 别名]
def run_pool():
    try:
        httpd = ThreadedTCPServer((State.config.get("proxy", "listen").strip(),
                                   State.config.getint("proxy", "port")), Proxy)
    except OSError as exc:
        if "attempt was made" in str(exc):
            print("Px failed to start - port in use")
        else:
            pprint(exc)
        return
    mainsock = httpd.socket
    print_banner()
    if hasattr(socket, "fromshare"):
        workers = State.config.getint("settings", "workers")
        for i in range(workers-1):
            (pipeout, pipein) = multiprocessing.Pipe()
            p = multiprocessing.Process(target=start_worker, args=(pipeout,))
            p.daemon = True
            p.start()
            while p.pid is None:
                time.sleep(1)
            pipein.send(mainsock.share(p.pid))
    serve_forever(httpd)
###
# Proxy detection 
开发者ID:genotrance,项目名称:px,代码行数:32,代码来源:px.py


示例4: remoteProcessServer

# 需要导入模块: import socket [as 别名]
# 或者: from socket import fromshare [as 别名]
def remoteProcessServer(cls, q):
        # Recreate socket from shared data
        sdata = q.get()
        message = q.get()
        s = socket.fromshare(sdata)
        s2, c = s.accept()
        # Send the message
        s2.sendall(message)
        s2.close()
        s.close() 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:14,代码来源:test_socket.py


示例5: testShareLength

# 需要导入模块: import socket [as 别名]
# 或者: from socket import fromshare [as 别名]
def testShareLength(self):
        data = self.serv.share(os.getpid())
        self.assertRaises(ValueError, socket.fromshare, data[:-1])
        self.assertRaises(ValueError, socket.fromshare, data+b"foo") 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:6,代码来源:test_socket.py


示例6: testShareLocal

# 需要导入模块: import socket [as 别名]
# 或者: from socket import fromshare [as 别名]
def testShareLocal(self):
        data = self.serv.share(os.getpid())
        s = socket.fromshare(data)
        try:
            self.compareSockets(self.serv, s)
        finally:
            s.close() 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:9,代码来源:test_socket.py


示例7: detach

# 需要导入模块: import socket [as 别名]
# 或者: from socket import fromshare [as 别名]
def detach(self):
            '''Get the socket.  This should only be called once.'''
            with _resource_sharer.get_connection(self._id) as conn:
                share = conn.recv_bytes()
                return socket.fromshare(share) 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:7,代码来源:resource_sharer.py



注:本文中的socket.fromshare方法示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。