使用线程上的阻塞 Socket 订阅 UPnP 事件

Question

我很想订阅 UPnP 设备（WeMo 运动传感器）上的事件。我首先向设备发送 HTTP 订阅请求，设备应该开始在指定地址上向我发送事件通知。那部分工作正常（除了我收到太多通知；即使状态没有改变，但对于不同的线程来说这是一个不同的问题）

如果我在单独的 python 进程上运行 keepListening 函数，一切正常。但是，当我将函数作为线程运行时，它不起作用；

import socket
import requests
from threading import Thread

def keepListening(): #running this function on a separate process works
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.settimeout(600)
    sock.bind(('192.168.10.231',1234))
    sock.listen(5)

    while 1:
        notification = ''

    try:
        conn, addr =  sock.accept()
        conn.setblocking(1)
        notification= conn.recv(1024) 
        conn.sendall(r'''HTTP/1.1 200 OK
Content-Type: text/plain

''')
    except Exception as er:
        print er

    print notification




x = Thread(target=keepListening)
x.start()

message = {
'CALLBACK': '<http://192.168.10.231:1234>',
'NT': 'upnp:event',
'TIMEOUT': 'Second-600',
'HOST': '192.168.10.159:49153'}

k = requests.request('SUBSCRIBE','http://192.168.10.159:49153/upnp/event/basicevent1',headers=message)
print k
# keep doing other important works

每个事件通知都必须回复一个 200 OK 回复，否则设备不会再发送通知；一个事实，我学会了艰难的方式。我有一个可能很愚蠢的疑问是，当在线程中运行时，与单独的进程相反，回复消息不会及时发送，因此设备不会再发送任何通知。 值得一提的是，即使我在线程中运行该函数，我确实收到订阅后的初始通知（设备必须根据 UPnP 协议强制在订阅后立即发送初始通知），但我没有收到进一步的通知（表明我的 200 OK 回复没有正确通过；不过我确实在 Wireshark 中看到了）

您知道在线程（而不是单独的进程）中运行函数可能会导致失败的区别吗？

谢谢。

Answer 1

我会假设，发生的事情是您最终在线程变为活动状态并开始在接口上侦听之前发送订阅请求。所以设备无法连接到socket。

Answer 2

几天前我得到了一个 wemo 运动传感器、开关和树莓派，所以我开始修修补补。

脚本订阅 wemo 设备的“binaryState”事件。 每次事件发生时，它都会打印出一个“警报”（您可以在那里做其他事情）。 250 秒后，它会更新订阅。

要使脚本适应您的需要，您必须更改 IP：

localIp : 你的电脑

remoteIp：wemo-sensor或switch的ip

我是 python 新手（3 天前开始），所以脚本可能需要一些修改，但它可以工作。

import socket
import threading
import requests

host = ''
port = 1234
localIp = '<http://192.168.1.32:1234>' # local IP of your computer 
remoteIp = '192.168.1.47:49153' # the ip of the wemo device 

global uid # stores the uuid of the event
uid = ''

class client(threading.Thread):
    def __init__(self, conn):
        super(client, self).__init__()
        self.conn = conn
        self.data = ""

    def run(self):
        global uid
        while True:
            self.data = self.data + self.conn.recv(1024)

            if self.data.endswith(u"\r\n"):
                print self.data # data from the wemo device

                uidPos = self.data.find("uuid")
                if uidPos != -1: # data contains the uuid of the event 
                    uid = self.data[uidPos+5:uidPos+41]

                if "<BinaryState>1</BinaryState>" in self.data:
                    print "ALERT ------------------------------------------Alert"      
                    # NOTIFICATION ! 

                if "/e:propertyset" in self.data:
                    self.conn.sendall('HTTP/1.1 200 OK\r\nContent-Type:text/html\r\n\r\n')
                    return 
                self.data = ""

    def send_msg(self,msg):
        self.conn.send(msg)

    def close(self):
        self.conn.close()

class connectionThread(threading.Thread):
    def __init__(self, host, port):
        super(connectionThread, self).__init__()
        try:
            self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.s.bind((host,port))
            self.s.listen(5)

        except socket.error:
            print 'Failed to create socket'
            sys.exit()
        self.clients = []

    def run(self):
        while True:
            print uid
            conn, address = self.s.accept()
            c = client(conn) 
            c.start()
            print '[+] Client connected: {0}'.format(address[0])

def main():
    get_conns = connectionThread(host, port)
    get_conns.start()
    print get_conns.clients
    while True:
        try:
            response = raw_input() 
        except KeyboardInterrupt:
            sys.exit()

def setCalback():
  global uid
  threading.Timer(250.0, setCalback).start()
  if uid == "": # no uuid set so we subscribe to the event
    eventSubscribe()
  else:         # uuid is set, so we renew the subsciption
    eventRefresh()


def eventSubscribe(): # subscribe to the wemo-event 
    message = {
        'CALLBACK': localIp,
        'NT': 'upnp:event',
        'TIMEOUT': 'Second-300',
        'HOST': remoteIp} 
    k = requests.request('SUBSCRIBE', "http://"+remoteIp+'/upnp/event/basicevent1',headers=message)
    print k


def eventRefresh() # refresh the subscription with the known uuid
    myuid = "uuid:"+uid
    message = {
        'SID': myuid,
        'TIMEOUT': 'Second-300',
        'HOST': remoteIp } 
    k = requests.request('SUBSCRIBE',"http://"+remoteIp+'/upnp/event/basicevent1',headers=message)
    print k


if __name__ == '__main__':
    threading.Timer(2.0, setCalback).start() # wait 2 sec. then subscribe to the service
    main()