大家好,又见面了,我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。
Jetbrains全系列IDE稳定放心使用
废话少说,直接上代码。服务端ssdp_server.py:
# -*- coding: utf-8 -*-
import socket
SSDP_ADDR = '239.255.255.250'
SSDP_PORT = 1900
SERVICE_NAME = 'my_service'
class Connection():
def __init__(self, s, data, addr):
self.__s = s
self.__data = data
self.__addr = addr
self.is_find_service = False
def handle_request(self):
if self.__data.startswith('M-SEARCH * HTTP/1.1\r\n'):
self.__handle_search()
elif self.__data.startswith('HTTP/1.1 200 OK\r\n'):
self.__handle_ok()
def __handle_search(self):
props = self.__parse_props(['HOST', 'MAN', 'ST', 'MX'])
if not props:
return
if props['HOST'] != '%s:%d' % (SSDP_ADDR, SSDP_PORT) \
or props['MAN'] != '"ssdp:discover"' \
or props['ST'] != 'ssdp:all':
return
print 'RECV: %s' % str(self.__data)
print 'ADDR: %s' % str(self.__addr)
response = 'HTTP/1.1 200 OK\r\nST: %s\r\n\r\n' % SERVICE_NAME
self.__s.sendto(response, self.__addr)
def __handle_ok(self):
props = self.__parse_props(['ST'])
if not props:
return
if props['ST'] != SERVICE_NAME:
return
print 'RECV: %s' % str(self.__data)
print 'ADDR: %s' % str(self.__addr)
print 'Find service!!!!'
self.is_find_service = True
def __parse_props(self, target_keys):
lines = self.__data.split('\r\n')
props = {}
for i in range(1, len(lines)):
if not lines[i]:
continue
index = lines[i].find(':')
if index == -1:
return None
props[lines[i][:index]] = lines[i][index + 1:].strip()
if not set(target_keys).issubset(set(props.keys())):
return None
return props
class SSDPServer():
def __init__(self):
self.__s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.__s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
local_ip = socket.gethostbyname(socket.gethostname())
any_ip = '0.0.0.0'
# 绑定到任意地址和SSDP组播端口上
self.__s.bind((any_ip, SSDP_PORT))
# INFO: 使用默认值
# self.__s.setsockopt(socket.SOL_IP, socket.IP_MULTICAST_TTL, 20)
# self.__s.setsockopt(socket.SOL_IP, socket.IP_MULTICAST_LOOP, 1)
# self.__s.setsockopt(socket.SOL_IP, socket.IP_MULTICAST_IF,
# socket.inet_aton(intf) + socket.inet_aton('0.0.0.0'))
# INFO: 添加到多播组
self.__s.setsockopt(socket.SOL_IP, socket.IP_ADD_MEMBERSHIP,
socket.inet_aton(SSDP_ADDR) + socket.inet_aton(local_ip))
self.local_ip = local_ip
def start(self):
while True:
data, addr = self.__s.recvfrom(2048)
conn = Connection(self.__s, data, addr)
conn.handle_request()
self.__s.setsockopt(socket.SOL_IP, socket.IP_DROP_MEMBERSHIP,
socket.inet_aton(SSDP_ADDR) + socket.inet_aton(self.local_ip))
self.__s.close()
if __name__ == '__main__':
port = SSDPServer()
port.start()
客户端ssdp_client.py:
# -*- coding: utf-8 -*-
import socket
import time
import select
import ssdp_server
SSDP_ADDR = '239.255.255.250'
SSDP_PORT = 1900
MS = 'M-SEARCH * HTTP/1.1\r\nHOST: %s:%d\r\nMAN: "ssdp:discover"\r\nMX: 2\r\nST: ssdp:all\r\n\r\n' \
% (SSDP_ADDR, SSDP_PORT)
class SSDPClient():
def __init__(self):
self.__s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# INFO: 若绑定,服务端收到的是固定的地址和端口号
self.__s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
local_ip = socket.gethostbyname(socket.gethostname())
self.__s.bind((local_ip, 50000))
def start(self):
self.__send_search()
while True:
reads, _, _ = select.select([self.__s], [], [], 5)
if reads:
data, addr = self.__s.recvfrom(2048)
conn = ssdp_server.Connection(self.__s, data, addr)
conn.handle_request()
if conn.is_find_service:
break
else: # timeout
self.__send_search()
self.__s.close()
def __send_search(self):
print "Sending M-SEARCH..."
# INFO: 发送到SSDP组播地址上
self.__s.sendto(MS, (SSDP_ADDR, SSDP_PORT))
if __name__ == '__main__':
port = SSDPClient()
port.start()
发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/183233.html原文链接:https://javaforall.cn
【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛
【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...