Python实现基于WebSocket的stomp协议调试助手工具分享

stomp协议很简单，但是搜遍网络竟没找到一款合适的客户端工具。大多数提供的都是客户端库的使用。可能是太简单了吧！可是即便这样，假如有一可视化的工具，将方便的对stomp协议进行抓包调试。网上类似MQTT的客户端工具有很多，但是stomp协议调试工具很少，这里使用Python和websocket实现stomp协议的调试工具，分享给有需要的小伙伴。

STOMP 协议简介

STOMP（Simple 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
NULL

COMMAND：表示操作类型，如 CONNECT, SEND, SUBSCRIBE 等。

header1:value1：头信息，用于传递额外的信息。

message-body：消息体，可选部分。

NULL：报文结束标志，用 \x00 表示。

基于 WebSocket 实现 STOMP

WebSocket 是一种在单个 TCP 连接上进行全双工通信的协议。通过 WebSocket，可以实现实时的数据交换。结合 STOMP 协议，可以构建高效的实时消息系统。

核心类设计

Stomp 类

Stomp 类负责管理与 STOMP 服务器的连接、订阅和发送消息等操作。

class Stomp:def __init__(self, host, sockjs=False, wss=True):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, username=None, passcode=None):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, id=None, ack='auto', callback=None):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(target=self.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, msg=None):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[i+1:len(lines)-1]).replace('\x00', '')return command, headers, bodydef connect(self, username=None, passcode=None):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, msg=message)def ack(self, message_id, subscription):headers = {'id': message_id, 'subscription': subscription}self._transmit(CMD_ACK, headers)

界面工具实现

tkinter是python自带的标准gui库，对于我们自己日常做一些小程序出来给自己使用是非常不错的。因为tkinter相比较其它强大的gui库（PyQT，WxPython等等）而言要简单、方便、学起来也容易得很多，所以用来造个小工具非常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 : 534117529@qq.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("820x560+500+150")  # 尺寸位置self.root.resizable(False, False)self.interface()Style(theme='pulse')self.isConnect = Falseself.client = Nonedef interface(self):""""界面编写位置"""# 操作区域self.fr1 = Frame(self.root)self.fr1.place(x=0, y=0, width=220, height=600)  # 区域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(image=self._img)self.about.place(x=65, y=0, width=80, height=80)pos = 80self.lb_server = Label(self.fr1, text='地址:', anchor="e", fg='red')self.lb_server.place(x=0, y=pos, width=50, height=35)self.txt_server = Text(self.fr1)self.txt_server.place(x=65, y=pos, width=155, height=28)self.txt_server.insert("1.0", "ws://localhost:15674/ws")  # WebSocket 地址self.lb_port = Label(self.fr1, text='clientID:', anchor="e", fg='red')self.lb_port.place(x=0, y=pos + 40, width=50, height=35)self.txt_id = Text(self.fr1)self.txt_id.place(x=65, y=pos + 40, width=155, height=28)self.txt_id.insert("1.0", "stomp-client")self.lb_user = Label(self.fr1, text='用户名:', anchor="e", fg='red')self.lb_user.place(x=0, y=pos + 80, width=50, height=35)self.txt_name = Text(self.fr1)self.txt_name.place(x=65, y=pos + 80, width=155, height=28)self.txt_name.insert("1.0", "guest")self.lb_pwd = Label(self.fr1, text='密码:', anchor="e", fg='red')self.lb_pwd.place(x=0, y=pos + 120, width=50, height=35)self.txt_pwd = Text(self.fr1)self.txt_pwd.place(x=65, y=pos + 120, width=155, height=28)self.txt_pwd.insert("1.0", "guest")self.var_bt1 = StringVar()self.var_bt1.set("连接")self.btn1 = Button(self.fr1, textvariable=self.var_bt1, command=self.btn_connect)self.btn1.place(x=170, y=pos + 160, width=50, height=30)self.lb_s = Label(self.fr1, text='订阅主题', bg="yellow", anchor='w')self.lb_s.place(x=5, y=340, width=90, height=28)self.txt_sub = Text(self.fr1)self.txt_sub.place(x=5, y=368, width=155, height=28)self.btn5 = Button(self.fr1, text='订阅', command=self.btn_sub)self.btn5.place(x=170, y=368, width=50, height=28)self.subitem = Listbox(self.fr1)self.subitem.place(x=5, y=402, width=215, height=85)self.subitem.bind("<Button-3>", self.on_right_click)# 文本区域self.fr2 = Frame(self.root)self.fr2.place(x=220, y=0, width=620, height=560)self.txt_rx = Text(self.fr2)self.txt_rx.place(relheight=0.6, relwidth=0.9, relx=0.05, rely=0.01)self.scrollbar = Scrollbar(self.txt_rx)self.scrollbar.pack(side=RIGHT, fill=Y)self.txt_rx.config(yscrollcommand=self.scrollbar.set)self.scrollbar.config(command=self.txt_rx.yview)self.txt_rx.bind("<Configure>", self.check_scrollbar)self.lb_t = Label(self.fr2, text='发布主题', bg="yellow", anchor='w')self.lb_t.place(relheight=0.04, relwidth=0.2, relx=0.05, rely=0.62)self.txt_topic = Text(self.fr2)self.txt_topic.place(relheight=0.05, relwidth=0.9, relx=0.05, rely=0.66)self.txt_tx = Text(self.fr2)self.txt_tx.place(relheight=0.15, relwidth=0.9, relx=0.05, rely=0.72)self.btn3 = Button(self.fr2, text='清空',command = self.txt_clr) #绑定清空方法self.btn4 = Button(self.fr2, text='保存',command=self.savefiles) #绑定保存方法self.btn3.place(relheight=0.06,relwidth=0.11,relx=0.05,rely=0.88)self.btn4.place(relheight=0.06,relwidth=0.11,relx=0.18,rely=0.88)self.btn6 = Button(self.fr2, text='发送', command=self.btn_send)self.btn6.place(relheight=0.06, relwidth=0.11, relx=0.84, rely=0.88)self.lb3 = Label(self.fr2, text='接收:0    发送:0', bg="yellow", anchor='w')self.lb3.place(relheight=0.05, relwidth=0.3, relx=0.045, rely=0.945)def check_scrollbar(self, *args):if self.txt_rx.yview() == (0.0, 1.0):self.scrollbar.pack_forget()else:self.scrollbar.place(RIGHT, fill=Y)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(text=timestr)  # 重新设置标签文本# 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,rx=0,tx=0):  #发送接收统计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，否则返回 False"""if json_str is None:return Falsetry:json.loads(json_str)return Trueexcept ValueError:return Falsedef appendTxt(self, msg, flag=None):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, sockjs=False, wss=False)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(f"Failed 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, indent=4, sort_keys=True, separators=(',', ': '), ensure_ascii=False)if(self.is_valid_json(body)):body = json.loads(str(body).replace("'", '"'))body = json.dumps(body, indent=4, sort_keys=True, separators=(',', ': '), ensure_ascii=False)self.appendTxt(f"Received 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) +',reason='+reason)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(destination=sub, ack='auto')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(destination=pub_topic,message=payload)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.1'class Stomp:def __init__(self, host, sockjs=False, wss=True):"""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 + "/websocket"protocol = "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,username=None,passcode=None):"""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,id=None,ack='auto',callback=None):"""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 destination"""self.dispatcher.send(destination, message)class Dispatcher:def __init__(self, stomp):"""The Dispatcher handles all network I/O and frame marshalling/unmarshalling"""self.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(target=self.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 WS"""print("<<< " + 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 out"""print(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 closed"""print("### 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 opened"""self.opened = Truedef _on_ping(self,ws,message):print("### ping ###")def _transmit(self, command, headers, msg=None):"""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 string"""lines = 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[i+1] == BYTE['NULL'] else  ''.join(lines[i+1:len(lines)-1])if body is not None:body = body.replace('\x00', '')else:body = Nonereturn command, headers, bodydef connect(self,username=None,passcode=None):"""Transmit a CONNECT frame"""headers = {}headers[HDR_HOST] = '/'headers[HDR_ACCEPT_VERSION] = VERSIONSheaders[HDR_HEARTBEAT] = '10000,10000'if 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 frame"""headers = {}# 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 frame"""headers = {}headers[HDR_DESTINATION] = destinationheaders[HDR_CONTENT_LENGTH] = str(len(message))self._transmit(CMD_SEND, headers, msg=message)def ack(self, message_id, subscription):"""Transmit an ACK frameACK 命令用于确认消息已成功处理当客户端接收到消息时，消息的头部会包含 message-id 字段。客户端需要从这个字段中提取 message_id在订阅消息时，客户端会指定一个 id，这个 id 就是 subscription"""headers = {}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, sockjs=False, wss=True)stomp.connect()stomp.subscribe(sub_topic, do_thing_a)time.sleep(2)stomp.send(send_topic, '{"name":"akshaye"}')
if __name__ == "__main__":main()