Source code for evennia.server.portal.mccp
"""
MCCP - Mud Client Compression Protocol
This implements the MCCP v2 telnet protocol as per
http://tintin.sourceforge.net/mccp/. MCCP allows for the server to
compress data when sending to supporting clients, reducing bandwidth
by 70-90%.. The compression is done using Python's builtin zlib
library. If the client doesn't support MCCP, server sends uncompressed
as normal. Note: On modern hardware you are not likely to notice the
effect of MCCP unless you have extremely heavy traffic or sits on a
terribly slow connection.
This protocol is implemented by the telnet protocol importing
mccp_compress and calling it from its write methods.
"""
import weakref
import zlib
# negotiations for v1 and v2 of the protocol
MCCP = bytes([86]) # b"\x56"
FLUSH = zlib.Z_SYNC_FLUSH
[docs]def mccp_compress(protocol, data):
"""
Handles zlib compression, if applicable.
Args:
data (str): Incoming data to compress.
Returns:
stream (binary): Zlib-compressed data.
"""
if hasattr(protocol, "zlib"):
return protocol.zlib.compress(data) + protocol.zlib.flush(FLUSH)
return data
[docs]class Mccp:
"""
Implements the MCCP protocol. Add this to a
variable on the telnet protocol to set it up.
"""
[docs] def __init__(self, protocol):
"""
initialize MCCP by storing protocol on
ourselves and calling the client to see if
it supports MCCP. Sets callbacks to
start zlib compression in that case.
Args:
protocol (Protocol): The active protocol instance.
"""
self.protocol = weakref.ref(protocol)
self.protocol().protocol_flags["MCCP"] = False
# ask if client will mccp, connect callbacks to handle answer
self.protocol().will(MCCP).addCallbacks(self.do_mccp, self.no_mccp)
[docs] def no_mccp(self, option):
"""
Called if client doesn't support mccp or chooses to turn it off.
Args:
option (Option): Option dict (not used).
"""
if hasattr(self.protocol(), "zlib"):
del self.protocol().zlib
self.protocol().protocol_flags["MCCP"] = False
self.protocol().handshake_done()
[docs] def do_mccp(self, option):
"""
The client supports MCCP. Set things up by
creating a zlib compression stream.
Args:
option (Option): Option dict (not used).
"""
self.protocol().protocol_flags["MCCP"] = True
self.protocol().requestNegotiation(MCCP, b"")
self.protocol().zlib = zlib.compressobj(9)
self.protocol().handshake_done()