persistent

README

Memory mapped files holding stream data, multi-producer, multi-consumer. Very Fast. Some ideas from Real-Logic Aeron.

• https://github.com/real-logic/aeron

Documentation

Index

• func OpenCreatePersistentStream(basePath string, partSize uint64, serialiser serialisation.StreamSerialiser) (ps api.PersistentStream, err error)
• type MmapStream
  • func MmapStreamCreate(baseFilename string, partSize uint64, ...) (s *MmapStream, err error)
  • func MmapStreamOpen(baseFilename string, serialiser serialisation.StreamSerialiser) (s *MmapStream, err error)
  • func (s *MmapStream) Close()
  • func (s *MmapStream) CloseFile() error
  • func (s *MmapStream) Consume(subscriberName string) api.Stream
  • func (s *MmapStream) Delete() error
  • func (s *MmapStream) Feed(elem interface{})
  • func (s *MmapStream) GetFirstPart() uint64
  • func (s *MmapStream) GetPartSize() uint64
  • func (s *MmapStream) GetPartsCount() uint64
  • func (s *MmapStream) GetRepHWM(repId int) uint64
  • func (s *MmapStream) GetReplicaOf() uint64
  • func (s *MmapStream) GetReplicatorIds() (reps []int)
  • func (s *MmapStream) GetUniqId() uint64
  • func (s *MmapStream) IsClosed() bool
  • func (s *MmapStream) Publish(uri string)
  • func (s *MmapStream) PullBySubId(subId int, timeOut api.WaitTimeOut, waitDuty api.WaitDuty) (elem interface{}, readAbsPos uint64, closed bool)
  • func (s *MmapStream) ReadSubRPos(subId int) uint64
  • func (s *MmapStream) ReplicatorIdForNameHost(name, host string) (repId, subId int, created bool)
  • func (s *MmapStream) Reset(subId int) uint64
  • func (s *MmapStream) SetRepHWM(repId int, HWM uint64)
  • func (s *MmapStream) SetReplicaOf(uniqId uint64)
  • func (s *MmapStream) SetSubRPos(subId int, absPos uint64)
  • func (s *MmapStream) SetWritePos(absPos uint64)
  • func (s *MmapStream) SubscriberIdForName(namedSubscriber string) int
  • func (s *MmapStream) WritePos() uint64

Functions

func OpenCreatePersistentStream

func OpenCreatePersistentStream(basePath string, partSize uint64, serialiser serialisation.StreamSerialiser) (ps api.PersistentStream, err error)

Types

type MmapStream

type MmapStream struct {
	// contains filtered or unexported fields
}

func MmapStreamCreate

func MmapStreamCreate(baseFilename string, partSize uint64, serialiser serialisation.StreamSerialiser) (s *MmapStream, err error)

func MmapStreamOpen

func MmapStreamOpen(baseFilename string, serialiser serialisation.StreamSerialiser) (s *MmapStream, err error)

Internal mmap Stream exported advanced usage, for streaming use the standard API: go_frank.PersistentStream

func (*MmapStream) Close

func (s *MmapStream) Close()

func (*MmapStream) CloseFile

func (s *MmapStream) CloseFile() error

func (*MmapStream) Consume

func (s *MmapStream) Consume(subscriberName string) api.Stream

func (*MmapStream) Delete

func (s *MmapStream) Delete() error

func (*MmapStream) Feed

func (s *MmapStream) Feed(elem interface{})

func (*MmapStream) GetFirstPart

func (s *MmapStream) GetFirstPart() uint64

func (*MmapStream) GetPartSize

func (s *MmapStream) GetPartSize() uint64

func (*MmapStream) GetPartsCount

func (s *MmapStream) GetPartsCount() uint64

func (*MmapStream) GetRepHWM

func (s *MmapStream) GetRepHWM(repId int) uint64

Gets Replica HighWaterMark

func (*MmapStream) GetReplicaOf

func (s *MmapStream) GetReplicaOf() uint64

func (*MmapStream) GetReplicatorIds

func (s *MmapStream) GetReplicatorIds() (reps []int)

func (*MmapStream) GetUniqId

func (s *MmapStream) GetUniqId() uint64

func (*MmapStream) IsClosed

func (s *MmapStream) IsClosed() bool

func (*MmapStream) Publish

func (s *MmapStream) Publish(uri string)

FIXME

func (*MmapStream) PullBySubId

func (s *MmapStream) PullBySubId(subId int, timeOut api.WaitTimeOut, waitDuty api.WaitDuty) (elem interface{}, readAbsPos uint64, closed bool)

TODO: this could return an []byte pointer to the mmap, so no copy; TODO: needs to differentiate between timeout and closed stream, to different things

func (*MmapStream) ReadSubRPos

func (s *MmapStream) ReadSubRPos(subId int) uint64

Gets Subscriber Read Position

func (*MmapStream) ReplicatorIdForNameHost

func (s *MmapStream) ReplicatorIdForNameHost(name, host string) (repId, subId int, created bool)

func (*MmapStream) Reset

func (s *MmapStream) Reset(subId int) uint64

func (*MmapStream) SetRepHWM

func (s *MmapStream) SetRepHWM(repId int, HWM uint64)

Sets Replica HighWaterMark

func (*MmapStream) SetReplicaOf

func (s *MmapStream) SetReplicaOf(uniqId uint64)

func (*MmapStream) SetSubRPos

func (s *MmapStream) SetSubRPos(subId int, absPos uint64)

Resets Subscriber Read Position to given AbsPos (advance: don't use, for replication purposes.)

func (*MmapStream) SetWritePos

func (s *MmapStream) SetWritePos(absPos uint64)

Resets Writer Position (advanced: don't use, for replication purposes.)

func (*MmapStream) SubscriberIdForName

func (s *MmapStream) SubscriberIdForName(namedSubscriber string) int

func (*MmapStream) WritePos

func (s *MmapStream) WritePos() uint64

Gets Writer Position