func ExpectedOffset(expectedOffset int64) MessageOption {
	return func(o *MessageOptions) {
		o.ExpectedOffset = expectedOffset
	}
}
ExpectedOffset set the value of expected offset after publishing the message.
This is required for optimistic concurrency control
package liftbridge
import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"hash/crc32"
	"sync"
	"time"
	proto "github.com/liftbridge-io/liftbridge-api/go"
	pb "github.com/golang/protobuf/proto"
)
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 asyncErrorFromProto(asyncError *proto.PublishAsyncError) error {
	if asyncError == nil {
		return nil
	}
	switch asyncError.Code {
	case proto.PublishAsyncError_NOT_FOUND:
		return ErrNoSuchPartition
	case proto.PublishAsyncError_READONLY:
		return ErrReadonlyPartition
	default:
		return errors.New(asyncError.Message)
	}
}
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
}
func (a *Ack) CorrelationID() string {
	return a.correlationID
}
func (a *Ack) AckPolicy() AckPolicy {
	return a.ackPolicy
}
func (a *Ack) ReceptionTimestamp() time.Time {
	return a.receptionTimestamp
}
func (a *Ack) CommitTimestamp() time.Time {
	return a.commitTimestamp
}
type Partitioner interface {
	Partition(stream string, key, value []byte, metadata *Metadata) int32
}
type keyPartitioner struct{}
func (k *keyPartitioner) Partition(stream string, key, value []byte, metadata *Metadata) int32 {
	if key == nil {
		key = []byte("")
	}
	partitions := metadata.PartitionCountForStream(stream)
	if partitions == 0 {
		return 0
	}
	return int32(hasher(key) % uint32(partitions))
}
type streamCounter struct {
	sync.Mutex
	count int32
}
type roundRobinPartitioner struct {
	sync.Mutex
	streamCounterMap map[string]*streamCounter
}
func newRoundRobinPartitioner() Partitioner {
	return &roundRobinPartitioner{
		streamCounterMap: make(map[string]*streamCounter),
	}
}
func (r *roundRobinPartitioner) Partition(stream string, key, value []byte, metadata *Metadata) int32 {
	partitions := metadata.PartitionCountForStream(stream)
	if partitions == 0 {
		return 0
	}
	r.Lock()
	counter, ok := r.streamCounterMap[stream]
	if !ok {
		counter = new(streamCounter)
		r.streamCounterMap[stream] = counter
	}
	r.Unlock()
	counter.Lock()
	count := counter.count
	counter.count++
	counter.Unlock()
	return count % partitions
}
type MessageOptions struct {
	Key []byte
	AckInbox string
	CorrelationID string
	AckPolicy AckPolicy
	Headers map[string][]byte
	Partitioner Partitioner
	Partition *int32
	ExpectedOffset int64
}
type MessageOption func(*MessageOptions)
func Key(key []byte) MessageOption {
	return func(o *MessageOptions) {
		o.Key = key
	}
}
func AckInbox(ackInbox string) MessageOption {
	return func(o *MessageOptions) {
		o.AckInbox = ackInbox
	}
}
func CorrelationID(correlationID string) MessageOption {
	return func(o *MessageOptions) {
		o.CorrelationID = correlationID
	}
}
func AckPolicyLeader() MessageOption {
	return func(o *MessageOptions) {
		o.AckPolicy = AckPolicy(proto.AckPolicy_LEADER)
	}
}
func AckPolicyAll() MessageOption {
	return func(o *MessageOptions) {
		o.AckPolicy = AckPolicy(proto.AckPolicy_ALL)
	}
}
func AckPolicyNone() MessageOption {
	return func(o *MessageOptions) {
		o.AckPolicy = AckPolicy(proto.AckPolicy_NONE)
	}
}
func Header(name string, value []byte) MessageOption {
	return func(o *MessageOptions) {
		o.Headers[name] = value
	}
}
func Headers(headers map[string][]byte) MessageOption {
	return func(o *MessageOptions) {
		for name, value := range headers {
			o.Headers[name] = value
		}
	}
}
func ToPartition(partition int32) MessageOption {
	return func(o *MessageOptions) {
		o.Partition = &partition
	}
}
func PartitionBy(partitioner Partitioner) MessageOption {
	return func(o *MessageOptions) {
		o.Partitioner = partitioner
	}
}
func PartitionByKey() MessageOption {
	return PartitionBy(partitionByKey)
}
func PartitionByRoundRobin() MessageOption {
	return PartitionBy(partitionByRoundRobin)
}