视频直播网站开发,网站流量怎么做的,四川省四川省住房和城乡建设厅网站,网站工信部备案流程stomp协议很简单#xff0c;但是搜遍网络竟没找到一款合适的客户端工具。大多数提供的都是客户端库的使用。可能是太简单了吧#xff01;可是即便这样#xff0c;假如有一可视化的工具#xff0c;将方便的对stomp协议进行抓包调试。网上类似MQTT的客户端工具有很多#xf… stomp协议很简单但是搜遍网络竟没找到一款合适的客户端工具。大多数提供的都是客户端库的使用。可能是太简单了吧可是即便这样假如有一可视化的工具将方便的对stomp协议进行抓包调试。网上类似MQTT的客户端工具有很多但是stomp协议调试工具很少这里使用Python和websocket实现stomp协议的调试工具分享给有需要的小伙伴。 STOMP 协议简介
STOMPSimple Text Oriented Messaging Protocol是一种简单的文本消息传递协议设计用于与消息中间件进行交互。它允许客户端通过多种编程语言与消息代理如ActiveMQ, RabbitMQ等进行通信。STOMP 协议的特点包括
简单协议设计简洁易于实现。
跨平台支持多种编程语言和操作系统。
灵活支持多种消息模式如发布/订阅、请求/响应等。 直接使用WebSocket或SockJS就很类似于使用TCP套接字来编写Web应用。因为没有高层级的线路协议因此就需要我们定义应用之间所发送消息的语义还需要确保连接的两端都能遵循这些语义。
就像HTTP在TCP套接字之上添加了请求-响应模型层一样STOMP在WebSocket之上提供了一个基于帧的线路格式frame-based wire format层用来定义消息的语义。
与HTTP请求和响应类似STOMP帧由命令、一个或多个头信息以及负载所组成。例如如下就是发送数据的一个STOMP帧 SEND
transaction:tx-0
destination:/app/marco
content-length:20{message:Marco!}它是一个基于帧的协议它的帧结构模仿了 HTTP。一个帧由命令、一组可选header和一个可选body组成。STOMP 是基于文本的但也允许传输二进制消息。STOMP 的默认编码是 UTF-8但它支持为消息主体指定备用编码。
STOMP 报文格式
STOMP 报文由命令、头信息和消息体组成格式如下
COMMAND
header1:value1
header2:value2message-body
NULLCOMMAND表示操作类型如 CONNECT, SEND, SUBSCRIBE 等。
header1:value1头信息用于传递额外的信息。
message-body消息体可选部分。
NULL报文结束标志用 \x00 表示。
基于 WebSocket 实现 STOMP
WebSocket 是一种在单个 TCP 连接上进行全双工通信的协议。通过 WebSocket可以实现实时的数据交换。结合 STOMP 协议可以构建高效的实时消息系统。
核心类设计
Stomp 类
Stomp 类负责管理与 STOMP 服务器的连接、订阅和发送消息等操作。
class Stomp:def __init__(self, host, sockjsFalse, wssTrue):self.url ws:// host if not wss else wss:// hostself.dispatcher Dispatcher(self)self.callback_registry {}self.on_error Noneself.on_connect Noneself.on_message Noneself.on_close Nonedef connect(self, usernameNone, passcodeNone):self.connected Falseself.dispatcher.connect(username, passcode)start_time time.time()timeout 10while not self.connected:if time.time() - start_time timeout:print(Connection timed out)return Falsetime.sleep(0.5)if self.on_connect:self.on_connect(self.connected)return self.connecteddef disconnect(self):self.dispatcher.ws.close()self.connected Falseif self.on_close:self.on_close()def subscribe(self, destination, idNone, ackauto, callbackNone):if callback:self.callback_registry[destination] callbackself.dispatcher.subscribe(destination, id, ack)def send(self, destination, message):self.dispatcher.send(destination, message)Dispatcher 类
Dispatcher 类负责处理 WebSocket 的连接、消息收发和帧的解析。
class Dispatcher:def __init__(self, stomp):self.stomp stompself.ws websocket.WebSocketApp(self.stomp.url)self.ws.on_open self._on_openself.ws.on_message self._on_messageself.ws.on_error self._on_errorself.ws.on_close self._on_closeself.ws.on_ping self._on_pingThread(targetself.ws.run_forever, kwargs{ping_interval: 10, ping_timeout: 8}).start()self.opened Falsewhile not self.opened:time.sleep(0.5)def _on_message(self, ws, message):print( message)command, headers, body self._parse_message(message)if command CONNECTED:self.stomp.connected Trueif command MESSAGE and headers[destination] in self.stomp.callback_registry:self.stomp.callback_registry[headers[destination]](body)if command ! and self.stomp.on_message:self.stomp.on_message(command, headers, body)def _on_error(self, ws, error):print(error)if self.stomp.on_error:self.stomp.on_error(error)def _on_close(self, ws, code, reason):print(### closed ###)if self.stomp.on_close:self.stomp.on_close(code, reason)def _on_open(self, ws):self.opened Truedef _on_ping(self, ws, message):print(### ping ###)def _transmit(self, command, headers, msgNone):lines [command BYTE[LF]]for key in headers:lines.append(key : headers[key] BYTE[LF])lines.append(BYTE[LF])if msg:lines.append(msg)lines.append(BYTE[NULL])frame .join(lines)print( frame)self.ws.send(frame)def _parse_message(self, frame):lines frame.split(BYTE[LF])command lines[0].strip()headers {}i 1while lines[i] ! :key, value lines[i].split(:)headers[key] valuei 1body None if i len(lines) - 1 else .join(lines[i1:len(lines)-1]).replace(\x00, )return command, headers, bodydef connect(self, usernameNone, passcodeNone):headers {HDR_HOST: /, HDR_ACCEPT_VERSION: VERSIONS, HDR_HEARTBEAT: 10000,10000}if username:headers[HDR_LOGIN] usernameif passcode:headers[HDR_PASSCODE] passcodeself._transmit(CMD_CONNECT, headers)def subscribe(self, destination, id, ack):headers {HDR_ID: id or str(uuid.uuid4()), CMD_ACK: ack, HDR_DESTINATION: destination}self._transmit(CMD_SUBSCRIBE, headers)def send(self, destination, message):headers {HDR_DESTINATION: destination, HDR_CONTENT_LENGTH: str(len(message))}self._transmit(CMD_SEND, headers, msgmessage)def ack(self, message_id, subscription):headers {id: message_id, subscription: subscription}self._transmit(CMD_ACK, headers)界面工具实现
tkinter是python自带的标准gui库对于我们自己日常做一些小程序出来给自己使用是非常不错的。因为tkinter相比较其它强大的gui库PyQTWxPython等等而言要简单、方便、学起来也容易得很多所以用来造个小工具非常nice但它做出来的界面不是很好看。
ttkbootstrap 介绍
ttkbootstrap 是一个基于 tkinter 和 ttk 的Python库它提供了一套现代化的主题和样式可以用于创建漂亮的图形用户界面GUI应用程序。它是基于 Bootstrap 框架的设计风格为 tkinter 应用程序提供了一致的外观和用户体验。
需要先安装依赖包
pip install ttkbootstrap
pip install -i https://pypi.doubanio.com/simple websocket-client # -*- coding: utf-8 -*-
# Time : 2023/09/17 12:49
# Author : yangyongzhen
# Email : 534117529qq.com
# File : stompclienttool.py
# Project : study
import time
import os
from tkinter.ttk import *
from tkinter import *
from datetime import datetime
from tkinter import messagebox
from ttkbootstrap import Style
#import stomp
import json
#import websocket
from PIL import Image, ImageTk
import stomp_wsglobal gui # 全局型式保存GUI句柄tx_cnt 0 # 发送条数统计
rx_cnt 0 # 接收条数统计class GUI:def __init__(self):self.root Tk()self.root.title(STOMP调试助手-author:blog.csdn.net/qq8864) # 窗口名称self.root.geometry(820x560500150) # 尺寸位置self.root.resizable(False, False)self.interface()Style(themepulse)self.isConnect Falseself.client Nonedef interface(self):界面编写位置# 操作区域self.fr1 Frame(self.root)self.fr1.place(x0, y0, width220, height600) # 区域1位置尺寸img_path os.path.join(os.path.dirname(__file__), me.png)img Image.open(img_path) # 替换为你的图片路径img img.resize((80, 80))self._img ImageTk.PhotoImage(img)self.about Label(self.fr1)self.about.image self._imgself.about.configure(imageself._img)self.about.place(x65, y0, width80, height80)pos 80self.lb_server Label(self.fr1, text地址:, anchore, fgred)self.lb_server.place(x0, ypos, width50, height35)self.txt_server Text(self.fr1)self.txt_server.place(x65, ypos, width155, height28)self.txt_server.insert(1.0, ws://localhost:15674/ws) # WebSocket 地址self.lb_port Label(self.fr1, textclientID:, anchore, fgred)self.lb_port.place(x0, ypos 40, width50, height35)self.txt_id Text(self.fr1)self.txt_id.place(x65, ypos 40, width155, height28)self.txt_id.insert(1.0, stomp-client)self.lb_user Label(self.fr1, text用户名:, anchore, fgred)self.lb_user.place(x0, ypos 80, width50, height35)self.txt_name Text(self.fr1)self.txt_name.place(x65, ypos 80, width155, height28)self.txt_name.insert(1.0, guest)self.lb_pwd Label(self.fr1, text密码:, anchore, fgred)self.lb_pwd.place(x0, ypos 120, width50, height35)self.txt_pwd Text(self.fr1)self.txt_pwd.place(x65, ypos 120, width155, height28)self.txt_pwd.insert(1.0, guest)self.var_bt1 StringVar()self.var_bt1.set(连接)self.btn1 Button(self.fr1, textvariableself.var_bt1, commandself.btn_connect)self.btn1.place(x170, ypos 160, width50, height30)self.lb_s Label(self.fr1, text订阅主题, bgyellow, anchorw)self.lb_s.place(x5, y340, width90, height28)self.txt_sub Text(self.fr1)self.txt_sub.place(x5, y368, width155, height28)self.btn5 Button(self.fr1, text订阅, commandself.btn_sub)self.btn5.place(x170, y368, width50, height28)self.subitem Listbox(self.fr1)self.subitem.place(x5, y402, width215, height85)self.subitem.bind(Button-3, self.on_right_click)# 文本区域self.fr2 Frame(self.root)self.fr2.place(x220, y0, width620, height560)self.txt_rx Text(self.fr2)self.txt_rx.place(relheight0.6, relwidth0.9, relx0.05, rely0.01)self.scrollbar Scrollbar(self.txt_rx)self.scrollbar.pack(sideRIGHT, fillY)self.txt_rx.config(yscrollcommandself.scrollbar.set)self.scrollbar.config(commandself.txt_rx.yview)self.txt_rx.bind(Configure, self.check_scrollbar)self.lb_t Label(self.fr2, text发布主题, bgyellow, anchorw)self.lb_t.place(relheight0.04, relwidth0.2, relx0.05, rely0.62)self.txt_topic Text(self.fr2)self.txt_topic.place(relheight0.05, relwidth0.9, relx0.05, rely0.66)self.txt_tx Text(self.fr2)self.txt_tx.place(relheight0.15, relwidth0.9, relx0.05, rely0.72)self.btn3 Button(self.fr2, text清空,command self.txt_clr) #绑定清空方法self.btn4 Button(self.fr2, text保存,commandself.savefiles) #绑定保存方法self.btn3.place(relheight0.06,relwidth0.11,relx0.05,rely0.88)self.btn4.place(relheight0.06,relwidth0.11,relx0.18,rely0.88)self.btn6 Button(self.fr2, text发送, commandself.btn_send)self.btn6.place(relheight0.06, relwidth0.11, relx0.84, rely0.88)self.lb3 Label(self.fr2, text接收:0 发送:0, bgyellow, anchorw)self.lb3.place(relheight0.05, relwidth0.3, relx0.045, rely0.945)def check_scrollbar(self, *args):if self.txt_rx.yview() (0.0, 1.0):self.scrollbar.pack_forget()else:self.scrollbar.place(RIGHT, fillY)def on_right_click(self, w):idx self.subitem.curselection()if idx ():returnselected_item self.subitem.get(idx)ret messagebox.askyesno(取消订阅, 取消订阅:\n selected_item)if ret:self.subitem.delete(idx)self.client.unsubscribe(selected_item)self.appendTxt(取消订阅: selected_item)def gettim(self):#获取时间 未用timestr time.strftime(%H:%M:%S) # 获取当前的时间并转化为字符串self.lb4.configure(texttimestr) # 重新设置标签文本# tim_str str(datetime.datetime.now()) \n# self.lb4[text] tim_str#self.lb3[text] 接收str(rx_cnt),发送str(tx_cnt)self.txt_rx.after(1000, self.gettim) # 每隔1s调用函数 gettime 自身获取时间 GUI自带的定时函数def txt_clr(self):#清空显示self.txt_rx.delete(0.0, end) # 清空文本框self.txt_tx.delete(0.0, end) # 清空文本框def tx_rx_cnt(self,rx0,tx0): #发送接收统计global tx_cntglobal rx_cntrx_cnt rxtx_cnt txself.lb3[text] 接收str(rx_cnt),发送str(tx_cnt)def savefiles(self): #保存日志TXT文本try:with open(log.txt,a) as file: #a方式打开 文本追加模式file.write(self.txt_rx.get(0.0,end))messagebox.showinfo(提示, 保存成功)except:messagebox.showinfo(错误, 保存日志文件失败)def log_callback(self,client, userdata, level, buf):print(buf)def is_valid_json(self,json_str):判断字符串是否是有效的 JSONArgs:json_str (str): 需要判断的字符串Returns:bool: 如果字符串是有效的 JSON则返回 True否则返回 Falseif json_str is None:return Falsetry:json.loads(json_str)return Trueexcept ValueError:return Falsedef appendTxt(self, msg, flagNone):current_t datetime.now()current_ current_t.strftime(%Y-%m-%d %H:%M:%S )self.txt_rx.insert(END, current_)self.txt_rx.insert(END, msg)self.txt_rx.insert(END, \n)self.txt_rx.see(END)self.txt_rx.update_idletasks()def connect(self, ws_url, user, password):# 将 ws_url 分解成 (host, port) 形式的元组if ws_url.startswith(ws://):ws_url ws_url[5:]elif ws_url.startswith(wss://):ws_url ws_url[6:]else:raise ValueError(Invalid WebSocket URL)self.client stomp_ws.Stomp(ws_url, sockjsFalse, wssFalse)self.client.on_connect self.on_connectself.client.on_message self.on_messageself.client.on_error self.on_errorself.client.on_close self.on_closeself.isConnect self.client.connect(user,password)return self.isConnectdef on_connect(self, rc):if rc True:print(Connected to Stomp Broker ok!\n)self.appendTxt(Connected to Stomp Broker ok!\n)self.var_bt1.set(断开)self.isConnect Trueelse:print(Failed to connect, return code %d\n, rc)self.appendTxt(fFailed to connect\n)self.isConnect Falsedef on_message(self, cmd,header, body):self.tx_rx_cnt(1,0)print(Received message: \n str(header))header json.loads(str(header).replace(, ))header json.dumps(header, indent4, sort_keysTrue, separators(,, : ), ensure_asciiFalse)if(self.is_valid_json(body)):body json.loads(str(body).replace(, ))body json.dumps(body, indent4, sort_keysTrue, separators(,, : ), ensure_asciiFalse)self.appendTxt(fReceived message:\n[Cmd]:{cmd}\n[Header]:\n{header}\n[Body]:\n{body}\n,RECV)def on_error(self, error):self.appendTxt(f发生错误: {error})def on_close(self,code,reason):self.isConnect Falseself.var_bt1.set(连接)self.subitem.delete(0, END)self.appendTxt(WebSocket连接已关闭,code str(code) ,reasonreason)def btn_connect(self): # 连接if self.var_bt1.get() 连接:server self.txt_server.get(1.0, END).strip()user self.txt_name.get(1.0, END).strip()psd self.txt_pwd.get(1.0, END).strip()ws_url server # WebSocket 地址print(f连接到 {ws_url}用户名: {user})self.appendTxt(f连接到 {ws_url}用户名: {user})if self.connect(ws_url, user, psd):self.var_bt1.set(断开)else:self.client.disconnect()self.var_bt1.set(连接)self.isConnect Falseself.appendTxt(断开连接)def btn_sub(self): # 订阅if self.isConnect:sub self.txt_sub.get(1.0, END).strip()self.client.subscribe(destinationsub, ackauto)self.appendTxt(f已订阅主题: {sub})self.subitem.insert(END, sub)else:messagebox.showinfo(提示, 服务器未连接!)def btn_send(self): # 发布if self.isConnect:pub_topic self.txt_topic.get(1.0, END).strip()payload self.txt_tx.get(1.0, END).strip()self.client.send(destinationpub_topic,messagepayload)self.appendTxt(f发布到 {pub_topic}: {payload})self.tx_rx_cnt(0,1)else:messagebox.showinfo(提示, 请连接服务器!)if __name__ __main__:print(Start...)gui GUI()gui.root.mainloop()print(End...)完整代码
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# author jenny
# datetime 2021/5/6 15:53
# File stomp_ws.py
import websocket
import time
from threading import Thread
import uuid
from constants import *
BYTE {LF: \x0A,NULL: \x00
}VERSIONS 1.0,1.1class Stomp:def __init__(self, host, sockjsFalse, wssTrue):Initialize STOMP communication. This is the high level API that is exposed to clients.Args:host: Hostnamesockjs: True if the STOMP server is sockjswss: True if communication is over SSL# websocket.enableTrace(True)ws_host host if sockjs is False else host /websocketprotocol ws:// if wss is False else wss://self.url protocol ws_hostprint(websocket url:self.url)self.dispatcher Dispatcher(self)# maintain callback registry for subscriptions - topic (str) vs callback (func)self.callback_registry {}self.on_error Noneself.on_connect Noneself.on_message Noneself.on_close Nonedef connect(self,usernameNone,passcodeNone):Connect to the remote STOMP server# set flag to falseself.connected False# attempt to connectself.dispatcher.connect(username,passcode)# wait until connectedstart_time time.time()timeout 10 # 10 secondswhile self.connected is False:if time.time() - start_time timeout:print(Connection timed out)return Falsetime.sleep(.50)if self.on_connect is not None:self.on_connect(self.connected)return self.connecteddef disconnect(self):Disconnect from the remote STOMP server self.dispatcher.ws.close()self.connected Falseif self.on_close is not None:self.on_close()def subscribe(self, destination,idNone,ackauto,callbackNone):Subscribe to a destination and supply a callback that should be executed when a message is received on that destination# create entry in registry against destinationif callback is not None:self.callback_registry[destination] callback# transmit subscribe frameself.dispatcher.subscribe(destination,id,ack)def send(self, destination, message):Send a message to a destinationself.dispatcher.send(destination, message)class Dispatcher:def __init__(self, stomp):The Dispatcher handles all network I/O and frame marshalling/unmarshallingself.stomp stomp#websocket.enableTrace(True) # 开启调试信息self.ws websocket.WebSocketApp(self.stomp.url)self.ws.ping_interval 30self.ws.ping_timeout 10# register websocket callbacksself.ws.on_open self._on_openself.ws.on_message self._on_messageself.ws.on_error self._on_errorself.ws.on_close self._on_closeself.ws.on_ping self._on_ping# run event loop on separate threadThread(targetself.ws.run_forever,kwargs{ping_interval: 10, ping_timeout: 8}).start()self.opened False# wait until connectedwhile self.opened is False:time.sleep(.50)def _on_message(self, ws, message):Executed when messages is received on WSprint( message)if len(message) 0:command, headers, body self._parse_message(message)# if connected, let Stomp knowif command CONNECTED:self.stomp.connected True# if message received, call appropriate callbackif command MESSAGE:# 检查字典中是否存在该主题的回调函数if headers[destination] in self.stomp.callback_registry:self.stomp.callback_registry[headers[destination]](body)# if message is acked, let Stomp knowif command CMD_ACK:print(ACK: headers[id])if command ! :if self.stomp.on_message is not None:self.stomp.on_message(command, headers, body)def _on_error(self, ws, error):Executed when WS connection errors outprint(error)if self.stomp.on_error is not None:self.stomp.on_error(error)def _on_close(self,ws,code,reason):Executed when WS connection is closedprint(### closed ###)if self.stomp.on_close is not None:self.stomp.on_close(code,reason)def _on_open(self, ws):Executed when WS connection is openedself.opened Truedef _on_ping(self,ws,message):print(### ping ###)def _transmit(self, command, headers, msgNone):Marshalls and transmits the frame# Contruct the framelines []lines.append(command BYTE[LF])# add headersfor key in headers:lines.append(key : headers[key] BYTE[LF])lines.append(BYTE[LF])# add message, if anyif msg is not None:lines.append(msg)# terminate with null octetlines.append(BYTE[NULL])frame .join(lines)# transmit over wsprint( frame)self.ws.send(frame)def _parse_message(self, frame):Returns:commandheadersbodyArgs:frame: raw frame stringlines frame.split(BYTE[LF])command lines[0].strip()headers {}# get all headersi 1while lines[i] ! :# get key, value from raw header(key, value) lines[i].split(:)headers[key] valuei 1# set body to None if there is no bodyif i len(lines) - 1:body None if lines[i1] BYTE[NULL] else .join(lines[i1:len(lines)-1])if body is not None:body body.replace(\x00, )else:body Nonereturn command, headers, bodydef connect(self,usernameNone,passcodeNone):Transmit a CONNECT frameheaders {}headers[HDR_HOST] /headers[HDR_ACCEPT_VERSION] VERSIONSheaders[HDR_HEARTBEAT] 10000,10000if username is not None:headers[HDR_LOGIN] usernameif passcode is not None:headers[HDR_PASSCODE] passcodeself._transmit(CMD_CONNECT, headers)def subscribe(self,destination,id,ack):Transmit a SUBSCRIBE frameheaders {}# TODO id should be auto generatedif id is None:id str(uuid.uuid4())headers[HDR_ID] idheaders[CMD_ACK] ackheaders[HDR_DESTINATION] destinationself._transmit(CMD_SUBSCRIBE, headers)def send(self, destination, message):Transmit a SEND frameheaders {}headers[HDR_DESTINATION] destinationheaders[HDR_CONTENT_LENGTH] str(len(message))self._transmit(CMD_SEND, headers, msgmessage)def ack(self, message_id, subscription):Transmit an ACK frameACK 命令用于确认消息已成功处理当客户端接收到消息时消息的头部会包含 message-id 字段。客户端需要从这个字段中提取 message_id在订阅消息时客户端会指定一个 id这个 id 就是 subscriptionheaders {}headers[id] message_idheaders[subscription] subscriptionself._transmit(CMD_ACK, headers)def do_thing_a(msg):print(MESSAGE: msg)def main(url,*sub_topic, **send_topic):stomp Stomp(url, sockjsFalse, wssTrue)stomp.connect()stomp.subscribe(sub_topic, do_thing_a)time.sleep(2)stomp.send(send_topic, {name:akshaye})
if __name__ __main__:main()测试使用
前提条件装有RabbitMQ并配置开启支持stomp协议。
工具下载
其他资源
https://github.com/jasonrbriggs/stomp.py
快速开始 | EMQX 企业版 4.3 文档
STOMP Over WebSocket
Fitten Code
https://github.com/rabbitmq/rabbitmq-server
https://www.rabbitmq.com/docs/install-windows#installer
python网络编程之websocket - 简书
【stomp实战】Stomp协议介绍和客户端的使用-CSDN博客
STOMP协议1.2_stomp1.2-CSDN博客
websocket_client教程Python中的WebSocket客户端实战-CSDN博客
