func (d *StreamMessage) SetStatus(code StatusCode) {
	d.StatusCode = code
}
SetStatus sets a status code for the message.
package protocol
import (
	"bytes"
	"context"
	"encoding/binary"
	"fmt"
	"io"
)
const StreamMessageHeaderSize uint32 = 3
const (
	MagicStreamReq MagicCode = 0xE4
	MagicStreamRes MagicCode = 0xE5
)
type StreamMessageHeader struct {
	Op         OpCode      
	ExtraLen   uint8       
	StatusCode StatusCode  
}
type StreamMessage struct {
	Header
	StreamMessageHeader
	extra  interface{}
	value  []byte
	buf    *bytes.Buffer
	conn   io.ReadWriteCloser
	cancel context.CancelFunc
}
func NewStreamMessage(opcode OpCode) *StreamMessage {
	return &StreamMessage{
		Header: Header{
			Magic:   MagicStreamReq,
			Version: Version1,
		},
		StreamMessageHeader: StreamMessageHeader{
			Op: opcode,
		},
	}
}
func ConvertToStreamMessage(msg EncodeDecoder, listenerID uint64) *StreamMessage {
	str := NewStreamMessage(OpStreamMessage)
	str.SetValue(msg.Buffer().Bytes())
	str.SetExtra(StreamMessageExtra{
		ListenerID: listenerID,
	})
	return str
}
func NewStreamMessageFromRequest(buf *bytes.Buffer) *StreamMessage {
	return &StreamMessage{
		Header: Header{
			Magic:         MagicStreamReq,
			Version:       Version1,
			MessageLength: uint32(buf.Len()),
		},
		StreamMessageHeader: StreamMessageHeader{},
		buf:                 buf,
	}
}
func (d *StreamMessage) Response(buf *bytes.Buffer) EncodeDecoder {
	msg := &StreamMessage{
		Header: Header{
			Magic:   MagicStreamRes,
			Version: Version1,
		},
		StreamMessageHeader: StreamMessageHeader{
			Op: d.Op,
		},
		buf: d.buf,
	}
	if buf != nil {
		msg.buf = buf
	} else {
		d.buf.Reset()
		msg.buf = d.buf
	}
	msg.MessageLength = uint32(d.buf.Len())
	return msg
}