func (a *Ack) CorrelationID() string {
	return a.correlationID
}
CorrelationID is the user-supplied value from the message.
package liftbridge
import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"hash/crc32"
	"sync"
	"time"
	pb "github.com/golang/protobuf/proto"
	proto "github.com/liftbridge-io/liftbridge-api/go"
)
var (
	envelopeMagicNumber    = []byte{0xB9, 0x0E, 0x43, 0xB4}
	envelopeMagicNumberLen = len(envelopeMagicNumber)
	partitionByKey         = new(keyPartitioner)
	partitionByRoundRobin  = newRoundRobinPartitioner()
	hasher                 = crc32.ChecksumIEEE
	crc32cTable            = crc32.MakeTable(crc32.Castagnoli)
)
const (
	envelopeProtoV0      = 0x00
	envelopeMinHeaderLen = 8
)
type msgType byte
const (
	msgTypePublish msgType = iota
	msgTypeAck
)
type AckPolicy int32
func (a AckPolicy) toProto() proto.AckPolicy {
	return proto.AckPolicy(a)
}
type Message struct {
	offset       int64
	key          []byte
	value        []byte
	partition    int32
	timestamp    time.Time
	stream       string
	subject      string
	replySubject string
	headers      map[string][]byte
}
func messageFromProto(wireMsg *proto.Message) *Message {
	if wireMsg == nil {
		return nil
	}
	msg := &Message{
		offset:       wireMsg.GetOffset(),
		key:          wireMsg.GetKey(),
		value:        wireMsg.GetValue(),
		partition:    wireMsg.GetPartition(),
		timestamp:    time.Unix(0, wireMsg.GetTimestamp()),
		stream:       wireMsg.GetStream(),
		subject:      wireMsg.GetSubject(),
		replySubject: wireMsg.GetReplySubject(),
		headers:      wireMsg.GetHeaders(),
	}
	return msg
}
func (m *Message) Offset() int64 {
	return m.offset
}
func (m *Message) Key() []byte {
	return m.key
}
func (m *Message) Value() []byte {
	return m.value
}
func (m *Message) Timestamp() time.Time {
	return m.timestamp
}
func (m *Message) Subject() string {
	return m.subject
}
func (m *Message) ReplySubject() string {
	return m.replySubject
}
func (m *Message) Headers() map[string][]byte {
	headers := make(map[string][]byte, len(m.headers))
	for key, value := range m.headers {
		headers[key] = value
	}
	return headers
}
func (m *Message) Stream() string {
	return m.stream
}
func (m *Message) Partition() int32 {
	return m.partition
}
type Ack struct {
	stream             string
	partitionSubject   string
	messageSubject     string
	offset             int64
	ackInbox           string
	correlationID      string
	ackPolicy          AckPolicy
	receptionTimestamp time.Time
	commitTimestamp    time.Time
}
func ackFromProto(wireAck *proto.Ack) *Ack {
	if wireAck == nil {
		return nil
	}
	ack := &Ack{
		stream:             wireAck.GetStream(),
		partitionSubject:   wireAck.GetPartitionSubject(),
		messageSubject:     wireAck.GetMsgSubject(),
		offset:             wireAck.GetOffset(),
		ackInbox:           wireAck.GetAckInbox(),
		correlationID:      wireAck.GetCorrelationId(),
		ackPolicy:          AckPolicy(wireAck.GetAckPolicy()),
		receptionTimestamp: time.Unix(0, wireAck.GetReceptionTimestamp()),
		commitTimestamp:    time.Unix(0, wireAck.GetCommitTimestamp()),
	}
	return ack
}
func (a *Ack) Stream() string {
	return a.stream
}
func (a *Ack) PartitionSubject() string {
	return a.partitionSubject
}
func (a *Ack) MessageSubject() string {
	return a.messageSubject
}
func (a *Ack) Offset() int64 {
	return a.offset
}
func (a *Ack) AckInbox() string {
	return a.ackInbox
}