mqtt 协议之 PINGREQ, PINGRESP

　　mqtt 协议里最简单的是 ping 协议吧 (心跳包)， ping 协议是已连接的客户端发往服务端, 告诉服务端，我还"活着"

PINGREQ - PING request

fixed header format.

bit	7	6	5	4	3	2	1	0
byte 1	Message Type (12)	DUP flag	QoS level	RETAIN
	1	1	0	0	x	x	x	x
byte 2	Remaining Length (0)
	0	0	0	0	0	0	0	0

no variable header

no payload

response: The response to a PINGREQ message is a PINGRESP message.

PINGRESP - PING response

fixed header

bit	7	6	5	4	3	2	1	0
byte 1	Message Type (13)	DUP flag	QoS level	RETAIN
	1	1	0	1	x	x	x	x
byte 2	Remaining Length (0)
	0	0	0	0	0	0	0	0

no variable header

no payload

------------------------------------------------------------------------ 华丽的分界线 ---------------------------------------

客户端会在一个心跳周期内发送一条PINGREQ消息到服务器端。两个字节，固定值。

服务器收到PINGREQ请求之后，会立即响应一个两个字节固定格式的PINGRESP消息。

周期定义在 心跳频率在CONNECT（连接包）可变头部“Keep Alive timer”中定义时间，单位为秒，无符号16位short表示。

ok ，上代码 ：

固定头部 FinedHeader

/// <summary>
/// Fixed header
/// </summary>
internal class FixedHeader
{
/// <summary>
/// Message type
/// </summary>
public MessageType MessageType { get; set; }

/// <summary>
/// DUP flag
/// </summary>
public bool Dup { get; set; }

/// <summary>
/// QoS flags
/// </summary>
public Qos Qos { get; set; }

/// <summary>
/// RETAIN 保持
/// </summary>
public bool Retain { get; set; }

/// <summary>
/// Remaining Length 剩余长度
/// 单个字节最大值：01111111，16进制：0x7F，10进制为127。
/// MQTT协议规定，第八位（最高位）若为1，则表示还有后续字节存在。
/// MQTT协议最多允许4个字节表示剩余长度。
/// 最大长度为：0xFF,0xFF,0xFF,0x7F，
/// 二进制表示为:11111111,11111111,11111111,01111111，十进制：268435455
/// </summary>
public int RemaingLength { get; set; }

public FixedHeader() { }

public FixedHeader(Stream stream)
{
if (stream.Length < 2)
throw new Exception("The supplied header is invalid. Header must be at least 2 bytes long.");

var byte1 = stream.ReadByte();
MessageType = (MessageType)((byte1 & 0xf0) >> 4);
Dup = ((byte1 & 0x08) >> 3) > 0;
Qos = (Qos)((byte1 & 0x06) >> 1);
Retain = (byte1 & 0x01) > 0;

//Remaining Length
//var byte2 = stream.ReadByte();
var lengthBytes = ReadLengthBytes(stream);
RemaingLength = CalculateLength(lengthBytes);
}

public void WriteTo(Stream stream)
{
var flags = (byte)MessageType << 4;
flags |= (Dup ? 1 : 0) << 3;
flags |= (byte)Qos << 1;
flags |= Retain ? 1 : 0;

stream.WriteByte((byte)flags);     //byte 1
if (RemaingLength == 0)         //byte 2
stream.WriteByte(0);
else
{
do
{
int digit = RemaingLength & 0x7f;
RemaingLength = RemaingLength >> 7;
if (RemaingLength > 0)
digit = digit | 0x80;
stream.WriteByte((byte)digit);
} while (RemaingLength > 0);
}
}

internal static byte[] ReadLengthBytes(Stream stream)
{
var lengthBytes = new List<byte>();

// read until we've got the entire size, or the 4 byte limit is reached
byte sizeByte;
int byteCount = 0;
do
{
sizeByte = (byte)stream.ReadByte();
lengthBytes.Add(sizeByte);
} while (++byteCount <= 4 && (sizeByte & 0x80) == 0x80);

return lengthBytes.ToArray();
}

internal static int CalculateLength(byte[] lengthBytes)
{
var remainingLength = 0;
var multiplier = 1;

foreach (var currentByte in lengthBytes)
{
remainingLength += (currentByte & 0x7f) * multiplier;
multiplier *= 0x80;
}

return remainingLength;
}
}

消息父类： Message

internal class Message
{
public FixedHeader FixedHeader { get; protected set; }

public Message()
{
}

public Message(MessageType messageType)
{
FixedHeader = new FixedHeader
{
MessageType = messageType
};
}

public virtual void WriteTo(Stream stream)
{
}

public static Message CreateFrom(byte[] buffer)
{
using (var stream = new MemoryStream(buffer))
{
return CreateFrom(stream);
}
}

public static Message CreateFrom(Stream stream)
{
var header = new FixedHeader(stream);
return CreateMessage(header, stream);
}

public static Message CreateMessage(FixedHeader header, Stream stream)
{
switch (header.MessageType)
{
case MessageType.CONNACK:
return new ConnAckMessage(header, stream);
case MessageType.DISCONNECT:
return null;
case MessageType.PINGREQ:
return new PingReqMessage();
case MessageType.PUBACK:
return new PublishAckMessage(header, stream);
case MessageType.PUBCOMP:
//return new MqttPubcompMessage(str, header);
case MessageType.PUBLISH:
//return new MqttPublishMessage(str, header);
case MessageType.PUBREC:
//return new MqttPubrecMessage(str, header);
case MessageType.PUBREL:
//return new MqttPubrelMessage(str, header);
case MessageType.SUBACK:
//return new MqttSubackMessage(str, header);
case MessageType.UNSUBACK:
//return new MqttUnsubackMessage(str, header);
case MessageType.PINGRESP:
return new PingRespMessage(header, stream);
case MessageType.UNSUBSCRIBE:
case MessageType.CONNECT:
case MessageType.SUBSCRIBE:
default:
throw new Exception("Unsupported Message Type");
}
}
}

两个枚举：

MessageType (消息类型)

Qos (服务质量等级)

[Flags]
public enum MessageType : byte
{
CONNECT     = 1,
CONNACK     = 2,
PUBLISH     = 3,
PUBACK      = 4,
PUBREC      = 5,
PUBREL      = 6,
PUBCOMP     = 7,
SUBSCRIBE   = 8,
SUBACK      = 9,
UNSUBSCRIBE = 10,
UNSUBACK    = 11,
PINGREQ     = 12,
PINGRESP    = 13,
DISCONNECT  = 14
}

/// <summary>
/// 服务质量等级
/// </summary>
[Flags]
public enum Qos : byte
{
/// <summary>
///     QOS Level 0 - Message is not guaranteed delivery. No retries are made to ensure delivery is successful.
/// </summary>
AtMostOnce = 0,

/// <summary>
///     QOS Level 1 - Message is guaranteed delivery. It will be delivered at least one time, but may be delivered
///     more than once if network errors occur.
/// </summary>
AtLeastOnce = 1,

/// <summary>
///     QOS Level 2 - Message will be delivered once, and only once. Message will be retried until
///     it is successfully sent..
/// </summary>
ExactlyOnce = 2,
}

ping 请求包: PingReqMessage

响应包: PingRespMessage

internal sealed class PingReqMessage : Message
{
public PingReqMessage()
: base(MessageType.PINGREQ)
{
}

public override void WriteTo(Stream stream)
{
FixedHeader.WriteTo(stream);
}
}

internal class PingRespMessage : Message
{
public PingRespMessage()
: base(MessageType.PINGRESP)
{
}

public PingRespMessage(FixedHeader header, Stream stream)
{
FixedHeader = header;
}
}

OK.