func DeserializeMessage(data []byte) (*Message, error) {
	serializable := new(SerializedMessage)
	if err := proto.Unmarshal(data, serializable); err != nil {
		return nil, err
	}
	msg := &Message{
		streamID:     MessageStreamID(serializable.GetStreamID()),
		prevStreamID: MessageStreamID(serializable.GetPrevStreamID()),
		origStreamID: MessageStreamID(serializable.GetOrigStreamID()),
		timestamp:    serializable.GetTimestamp(),
	}
	if msgData := serializable.GetData(); msgData != nil {
		msg.data.payload = msgData.GetData()
		metaGob := msgData.GetMetadata()
		if len(metaGob) > 0 {
			meta := NewMetadata()
			decoder := gob.NewDecoder(bytes.NewReader(metaGob))
			if err := decoder.Decode(&meta); err != nil {
				return nil, err
			}
			msg.data.metadata = meta
		}
	}
	if msgOrigData := serializable.GetOriginal(); msgOrigData != nil {
		msg.orig = new(MessageData)
		msg.orig.payload = msgOrigData.GetData()
		metaGob := msgOrigData.GetMetadata()
		if len(metaGob) > 0 {
			meta := NewMetadata()
			decoder := gob.NewDecoder(bytes.NewReader(metaGob))
			if err := decoder.Decode(&meta); err != nil {
				return nil, err
			}
			msg.orig.metadata = meta
		}
	}
	return msg, nil
}
DeserializeMessage generates a message from a byte array produced by
Message.Serialize. Please note that the payload is restored but the original
data is not. As of this FreezeOriginal can be called again after this call.
package core
import (
	"bytes"
	"encoding/gob"
	"time"
	"github.com/trivago/tgo/tcontainer"
	"github.com/golang/protobuf/proto"
)
type MessageData struct {
	payload  []byte
	metadata tcontainer.MarshalMap
}
type Message struct {
	data         MessageData
	orig         *MessageData
	streamID     MessageStreamID
	prevStreamID MessageStreamID
	origStreamID MessageStreamID
	source       MessageSource
	timestamp    int64
}
func NewMessage(source MessageSource, data []byte, metadata tcontainer.MarshalMap, streamID MessageStreamID) *Message {
	msg := &Message{
		source:       source,
		streamID:     streamID,
		origStreamID: streamID,
		timestamp:    time.Now().UnixNano(),
	}
	msg.data.payload = getPayloadCopy(data)
	if metadata != nil && len(metadata) > 0 {
		msg.data.metadata = metadata
	}
	return msg
}
func getPayloadCopy(data []byte) (buffer []byte) {
	buffer = make([]byte, len(data))
	copy(buffer, data)
	return
}
func (msg *Message) GetCreationTime() time.Time {
	return time.Unix(0, msg.timestamp)
}
func (msg *Message) GetStreamID() MessageStreamID {
	return msg.streamID
}
func (msg *Message) GetOrigStreamID() MessageStreamID {
	return msg.origStreamID
}
func (msg *Message) GetPrevStreamID() MessageStreamID {
	return msg.prevStreamID
}
func (msg *Message) GetRouter() Router {
	return StreamRegistry.GetRouterOrFallback(msg.GetStreamID())
}
func (msg *Message) GetPrevRouter() Router {
	return StreamRegistry.GetRouterOrFallback(msg.GetPrevStreamID())
}
func (msg *Message) GetOrigRouter() Router {
	return StreamRegistry.GetRouterOrFallback(msg.GetOrigStreamID())
}
func (msg *Message) SetStreamID(streamID MessageStreamID) {
	msg.prevStreamID = msg.streamID
	msg.streamID = streamID
}
func (msg *Message) SetlStreamIDAsOriginal(streamID MessageStreamID) {
	msg.SetStreamID(streamID)
	msg.origStreamID = streamID
}
func (msg *Message) GetSource() MessageSource {
	return msg.source
}
func (msg *Message) String() string {
	return string(msg.data.payload)
}
func (msg *Message) GetPayload() []byte {
	return msg.data.payload
}
func (msg *Message) GetMetadata() tcontainer.MarshalMap {
	if msg.data.metadata == nil {
		msg.data.metadata = NewMetadata()
	}
	return msg.data.metadata
}
func (msg *Message) TryGetMetadata() tcontainer.MarshalMap {
	return msg.data.metadata
}
func (msg *Message) StorePayload(data []byte) {
	if len(data) <= cap(msg.data.payload) {
		msg.data.payload = msg.data.payload[:len(data)]
		copy(msg.data.payload, data)
	} else {
		msg.data.payload = data
	}
}
func (msg *Message) Clone() *Message {
	clone := *msg
	clone.data.payload = make([]byte, len(msg.data.payload))
	copy(clone.data.payload, msg.data.payload)
	if clone.data.metadata != nil {
		clone.data.metadata = msg.data.metadata.Clone()
	}
	return &clone
}
func (msg *Message) CloneOriginal() *Message {
	if msg.orig == nil {
		msg.FreezeOriginal()
	}
	clone := *msg
	clone.data.payload = make([]byte, len(msg.orig.payload))
	copy(clone.data.payload, msg.orig.payload)
	if msg.orig.metadata != nil {
		clone.data.metadata = msg.orig.metadata.Clone()
	} else {
		clone.data.metadata = nil
	}
	clone.SetStreamID(msg.origStreamID)
	return &clone
}
func (msg *Message) FreezeOriginal() {
	if msg.orig != nil {
		return
	}
	var metadata tcontainer.MarshalMap
	if msg.data.metadata != nil {
		metadata = msg.data.metadata.Clone()
	}
	msg.orig = &MessageData{
		payload:  getPayloadCopy(msg.data.payload),
		metadata: metadata,
	}
}
func (msg *Message) Serialize() ([]byte, error) {
	metaBuffer := bytes.NewBuffer([]byte{})
	if len(msg.data.metadata) > 0 {
		encoder := gob.NewEncoder(metaBuffer)
		encoder.Encode(msg.data.metadata)
	}
	serializable := &SerializedMessage{
		StreamID:     proto.Uint64(uint64(msg.GetStreamID())),
		PrevStreamID: proto.Uint64(uint64(msg.GetPrevStreamID())),
		OrigStreamID: proto.Uint64(uint64(msg.GetOrigStreamID())),
		Timestamp:    proto.Int64(msg.timestamp),
		Data: &SerializedMessageData{
			Data:     msg.data.payload,
			Metadata: metaBuffer.Bytes(),
		},
	}
	if msg.orig != nil {
		origMetaBuffer := bytes.NewBuffer([]byte{})
		if len(msg.orig.metadata) > 0 {
			origEncoder := gob.NewEncoder(origMetaBuffer)
			origEncoder.Encode(msg.orig.metadata)
		}
		serializable.Original = &SerializedMessageData{
			Data:     msg.orig.payload,
			Metadata: origMetaBuffer.Bytes(),
		}
	}
	return proto.Marshal(serializable)
}