Asserting buffer contents across multiple readers.
diff --git a/example/example_consumer.go b/example/example_consumer.go
index aff7e79..d9ca62a 100644
--- a/example/example_consumer.go
+++ b/example/example_consumer.go
@@ -7,8 +7,6 @@
"github.com/smartystreets/go-disruptor"
)
-const Mod = 1000000 * 10 // 1 million * N
-
func consume0(reader *disruptor.SimpleReader) {
for {
reader.Receive()
@@ -22,7 +20,7 @@
if remaining >= 0 {
for remaining >= 0 {
- if sequence%Mod == 0 {
+ if sequence%ReportingFrequency == 0 {
finished := time.Now()
fmt.Println(sequence, finished.Sub(started))
started = time.Now()
diff --git a/example/example_consumer_handler.go b/example/example_consumer_handler.go
index 7de4b7e..7823b58 100644
--- a/example/example_consumer_handler.go
+++ b/example/example_consumer_handler.go
@@ -16,7 +16,7 @@
}
func (this *ExampleConsumerHandler) Consume(sequence, remaining int64) {
- if sequence%Mod == 0 {
+ if sequence%ReportingFrequency == 0 {
finished := time.Now()
fmt.Println(sequence, finished.Sub(this.started))
this.started = time.Now()
diff --git a/example/main.go b/example/main.go
index 9299264..b1164c9 100644
--- a/example/main.go
+++ b/example/main.go
@@ -6,9 +6,12 @@
"github.com/smartystreets/go-disruptor"
)
-const MaxConsumersPerGroup = 1
-const MaxConsumerGroups = 1
-const ItemsToPublish = 1
+const (
+ MaxConsumersPerGroup = 1
+ MaxConsumerGroups = 1
+ ItemsToPublish = 4
+ ReportingFrequency = 1000000 * 100 // 1 million * N
+)
func main() {
runtime.GOMAXPROCS(MaxConsumerGroups*MaxConsumersPerGroup + 1)
@@ -36,17 +39,19 @@
reader := disruptor.NewReader(upstreamBarrier, writerCursor, readerCursor)
// constant time regardless of the number of items
- go consume0(disruptor.NewSimpleReader(reader, NewExampleConsumerHandler()))
+ //go consume0(disruptor.NewSimpleReader(reader, NewExampleConsumerHandler()))
// wildly sporadic latency for single-item publish, e.g. 2 seconds, 65 ms, etc.
// faster for 2-3+ items per publish
- //go consume1(reader)
+ // go consume1(reader)
- // if group == 0 {
- // go consume1(reader)
- // } else if group == 1 {
- // go consume2(reader)
- // }
+ if group == 0 {
+ go consume1(reader)
+ } else if group == 1 {
+ go consume2(reader)
+ } else {
+ panic("only two consumer groups currently supported.")
+ }
}
return disruptor.NewBarrier(readerCursors...)
