__ __ __ __ _____ _ _ _____ _ _ _
| \/ | \ \ / / | __ \ (_) | | / ____| | | | |
| \ / |_ __\ V / | |__) | __ ___ ____ _| |_ ___ | (___ | |__ ___| | |
| |\/| | '__|> < | ___/ '__| \ \ / / _` | __/ _ \ \___ \| '_ \ / _ \ | |
| | | | |_ / . \ | | | | | |\ V / (_| | || __/ ____) | | | | __/ | |
|_| |_|_(_)_/ \_\ |_| |_| |_| \_/ \__,_|\__\___| |_____/|_| |_|\___V 2.1
if you need WebShell for Seo everyday contact me on Telegram
Telegram Address : @jackleet
For_More_Tools:
from __future__ import annotations
from typing import Callable
PUB_OP = "PUB"
HPUB_OP = "HPUB"
SUB_OP = "SUB"
UNSUB_OP = "UNSUB"
_CRLF_ = "\r\n"
Command = Callable[..., bytes]
def pub_cmd(subject, reply, payload) -> bytes:
return f"{PUB_OP} {subject} {reply} {len(payload)}{_CRLF_}".encode() + payload + _CRLF_.encode()
def hpub_cmd(subject, reply, hdr, payload) -> bytes:
hdr_len = len(hdr)
total_size = len(payload) + hdr_len
return f"{HPUB_OP} {subject} {reply} {hdr_len} {total_size}{_CRLF_}".encode() + hdr + payload + _CRLF_.encode()
def sub_cmd(subject, queue, sid) -> bytes:
return f"{SUB_OP} {subject} {queue} {sid}{_CRLF_}".encode()
def unsub_cmd(sid, limit) -> bytes:
limit_s = "" if limit == 0 else f"{limit}"
return f"{UNSUB_OP} {sid} {limit_s}{_CRLF_}".encode()