broker

Documentation

Index

• func DividePeerMessages(nrPeers int, numMessages uint) []uint
• func GenerateName() string
• type MQSettings
• type PeerOperations
• type Publisher
  • func (publisher *Publisher) StartPublishing(nrPeers int)
• type Result
• type Results
  • func (results *Results) PrintResults()
• type Subscriber
  • func (subscriber *Subscriber) StartSubscribing(nrPeers int)

Functions

func DividePeerMessages

func DividePeerMessages(nrPeers int, numMessages uint) []uint

func GenerateName

func GenerateName() string

Types

type MQSettings

type MQSettings struct {
	BrokerName  string
	BrokerHost  string
	BrokerPort  string
	Username    string
	Password    string
	MessageSize uint64
	NumMessages uint
	Producers   uint
	Consumers   uint
	QueueType   string
	ClusterID   string
	Topic       string
}

type PeerOperations

type PeerOperations interface {
	ReceiveMessage() ([]byte, error)
	SetupPublishRoutine()
	SendChannel() chan<- []byte
	ErrorChannel() <-chan error
	DoneChannel()
}

type Publisher

type Publisher struct {
	PeerOperations
	Id                       int
	NrMessagesToPublish      uint
	MessageSize              uint64
	Results                  *Result
	SyncMutex                *sync.Mutex
	SyncCond                 *sync.Cond
	NrReadyPeers             *int
	NrDonePeers              *int
	SubNrConsumedMessagesArr []*uint
	SubscriberDoneArr        []*bool
	NrPublishers             int
}

func (*Publisher) StartPublishing

func (publisher *Publisher) StartPublishing(nrPeers int)

type Result

type Result struct {
	PeerID     int
	Duration   float32
	Throughput float32
	Latency    *latencyResults
	Err        string
}

type Results

type Results struct {
	PublisherResults  []*Result
	SubscriberResults []*Result
}

func (*Results) PrintResults

func (results *Results) PrintResults()

type Subscriber

type Subscriber struct {
	PeerOperations
	Id                  int
	NrMessagesToConsume uint
	MessageSize         int64
	HasStarted          bool
	Started             int64
	Stopped             int64
	Results             *Result
	SyncMutex           *sync.Mutex
	SyncCond            *sync.Cond
	NrReadyPeers        *int
	NrDonePeers         *int
	NrMessagesConsumed  *uint
	SubscriberDone      *bool
}

func (*Subscriber) StartSubscribing

func (subscriber *Subscriber) StartSubscribing(nrPeers int)