Merge pull request #1 from alexsniffin/update-interface

Upgrade to Go 1.19 and API to support generics

Author: alexsniffin

.golangci.yml

@@ -10,4 +10,5 @@ output:
   print-issued-lines: true
   print-linter-name: true
   uniq-by-line: true
-  new: false
+  new: false
+

.travis.yml

@@ -1,9 +1,8 @@
-dist: bionic
+dist: jammy
 
 services:
   - docker
 
 script:
-  - docker build --build-arg VERSION=1.13 .
-  - docker build --build-arg VERSION=1.14 .
-  - docker build --build-arg VERSION=1.15 .
+  - docker build --build-arg VERSION=1.18 .
+  - docker build --build-arg VERSION=1.19 .

Dockerfile

@@ -14,7 +14,7 @@ COPY . $GOPATH/src/github.com/alexsniffin/gosd.git/
 WORKDIR $GOPATH/src/github.com/alexsniffin/gosd.git/
 
 # Pull dependencies
-RUN curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v1.31.0
+RUN curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin
 
 # Run quality check
 RUN golangci-lint run --timeout 10m0s -v --build-tags mus -c .golangci.yml \

README.md

@@ -7,12 +7,12 @@
 go-schedulable-dispatcher (gosd), is a library for scheduling when to dispatch a message to a channel.
 
 ## Implementation
-The implementation provides an ease-of-use API with both an ingress (ingest) channel and egress (dispatch) channel. Messages are ingested and processed into a heap based priority queue for dispatching. At most two separate goroutines are used, one for processing of messages from both the ingest channel and heap then the other as a timer. Order is not guaranteed by default when messages have the same scheduled time, but can be changed through the config. By guaranteeing order, performance will be slightly worse. If strict-ordering isn't critical to your application, it's recommended to keep the default setting.
+The implementation provides an interactive API to handle scheduling messages with a dispatcher. Messages are ingested and processed into a heap based priority queue. Order is not guaranteed by default when messages have the same scheduled time, but can be changed through the config. By guaranteeing order, performance will be slightly worse. If strict-ordering isn't critical to your application, it's recommended to keep the default setting.
 
 ## Example
 ```go
 // create instance of dispatcher
-dispatcher, err := gosd.NewDispatcher(&gosd.DispatcherConfig{
+dispatcher, err := gosd.NewDispatcher[string](&gosd.DispatcherConfig{
     IngressChannelSize:  100,
     DispatchChannelSize: 100,
     MaxMessages:         100,
@@ -24,17 +24,15 @@ checkErr(err)
 go dispatcher.Start()
 
 // schedule a message
-dispatcher.IngressChannel() <- &gosd.ScheduledMessage{
+dispatcher.IngressChannel() <- &gosd.ScheduledMessage[string]{
     At:      time.Now().Add(1 * time.Second),
     Message: "Hello World in 1 second!",
 }
 
 // wait for the message
 msg := <-dispatcher.DispatchChannel()
 
-// type assert
-msgStr := msg.(string)
-fmt.Println(msgStr)
+fmt.Println(msg)
 // Hello World in 1 second!
 
 // shutdown without deadline
@@ -44,13 +42,24 @@ dispatcher.Shutdown(context.Background(), false)
 More examples under [examples](examples).
 
 ## Benchmarking
-Tested with Intel Core i7-8700K CPU @ 3.70GHz, DDR4 RAM and 1000 messages per iteration.
+Tested with Go 1.19 and 1000 messages per iteration.
 ```
-Benchmark_integration_unordered-12               	     142	   8654906 ns/op
-Benchmark_integration_unorderedSmallBuffer-12    	     147	   9503403 ns/op
-Benchmark_integration_unorderedSmallHeap-12      	     122	   8860732 ns/op
-Benchmark_integration_ordered-12                 	      96	  13354174 ns/op
-Benchmark_integration_orderedSmallBuffer-12      	     121	  10115702 ns/op
-Benchmark_integration_orderedSmallHeap-12        	     129	  10441857 ns/op
-Benchmark_integration_orderedSameTime-12        	     99	   	  12575961 ns/op
+goos: windows
+goarch: amd64
+pkg: github.com/alexsniffin/gosd/v2
+cpu: Intel(R) Core(TM) i7-8700K CPU @ 3.70GHz
+Benchmark_integration_unordered
+Benchmark_integration_unordered-12                           307           3690528 ns/op
+Benchmark_integration_unorderedSmallBuffer
+Benchmark_integration_unorderedSmallBuffer-12                274           4120104 ns/op
+Benchmark_integration_unorderedSmallHeap
+Benchmark_integration_unorderedSmallHeap-12                  348           3452703 ns/op
+Benchmark_integration_ordered
+Benchmark_integration_ordered-12                             135           8650709 ns/op
+Benchmark_integration_orderedSmallBuffer
+Benchmark_integration_orderedSmallBuffer-12                  207           5867338 ns/op
+Benchmark_integration_orderedSmallHeap
+Benchmark_integration_orderedSmallHeap-12                    350           3592990 ns/op
+Benchmark_integration_orderedSameTime
+Benchmark_integration_orderedSameTime-12                     133           8909311 ns/op
 ```

delayer.go

@@ -4,9 +4,9 @@ import (
 	"time"
 )
 
-type delayer interface {
+type delayer[T any] interface {
 	stop(drain bool)
-	wait(msg *ScheduledMessage)
+	wait(msg *ScheduledMessage[T])
 	available() bool
 }
 
@@ -17,31 +17,33 @@ const (
 	waiting
 )
 
-type delay struct {
-	state delayState
+type delay[T any] struct {
+	state delayState // nolint:unused
 
 	idleChannel   chan<- bool
-	egressChannel chan<- interface{}
+	egressChannel chan<- T
 	cancelChannel chan bool
 }
 
-func newDelay(egressChannel chan<- interface{}, idleChannel chan<- bool) *delay {
-	return &delay{
+func newDelay[T any](egressChannel chan<- T, idleChannel chan<- bool) *delay[T] {
+	return &delay[T]{
 		idleChannel:   idleChannel,
 		egressChannel: egressChannel,
 		cancelChannel: make(chan bool, 1),
 	}
 }
 
 // stop sends a cancel signal to the current timer process.
-func (d *delay) stop(drain bool) {
+// nolint:unused
+func (d *delay[T]) stop(drain bool) {
 	if d.state == waiting {
 		d.cancelChannel <- drain
 	}
 }
 
 // wait will create a timer based on the time from `msg.At` and dispatch the message to the egress channel asynchronously.
-func (d *delay) wait(msg *ScheduledMessage) {
+// nolint:unused
+func (d *delay[T]) wait(msg *ScheduledMessage[T]) {
 	d.state = waiting
 	curTimer := time.NewTimer(time.Until(msg.At))
 
@@ -69,6 +71,7 @@ func (d *delay) wait(msg *ScheduledMessage) {
 }
 
 // available returns whether the delay is able to accept a new message to wait on.
-func (d *delay) available() bool {
+// nolint
+func (d *delay[T]) available() bool {
 	return d.state == idle
 }

delayer_test.go

@@ -21,7 +21,7 @@ func Test_delay_stop(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			d := &delay{
+			d := &delay[any]{
 				state:         tt.fields.state,
 				egressChannel: tt.fields.egressChannel,
 				cancelChannel: tt.fields.cancelChannel,
@@ -42,17 +42,17 @@ func Test_delay_wait(t *testing.T) {
 		cancelChannel chan bool
 	}
 	type args struct {
-		msg *ScheduledMessage
+		msg *ScheduledMessage[any]
 	}
 	tests := []struct {
 		name            string
 		fields          fields
 		args            args
-		customAssertion func(fields, *delay)
+		customAssertion func(fields, *delay[any])
 	}{
 		{"egressMessage", fields{
 			egressChannel: make(chan interface{}),
-			idleChannel:   make(chan bool)}, args{msg: &ScheduledMessage{At: time.Now()}}, func(f fields, d *delay) {
+			idleChannel:   make(chan bool)}, args{msg: &ScheduledMessage[any]{At: time.Now()}}, func(f fields, d *delay[any]) {
 			if d.state != waiting {
 				t.Errorf("wait() unexpected state = %+v, want Waiting", d.state)
 			}
@@ -68,7 +68,7 @@ func Test_delay_wait(t *testing.T) {
 		}},
 		{"cancelMessage", fields{
 			cancelChannel: make(chan bool, 1),
-			idleChannel:   make(chan bool)}, args{msg: &ScheduledMessage{At: time.Now().Add(10 + time.Second)}}, func(f fields, d *delay) {
+			idleChannel:   make(chan bool)}, args{msg: &ScheduledMessage[any]{At: time.Now().Add(10 + time.Second)}}, func(f fields, d *delay[any]) {
 			if d.state != waiting {
 				t.Errorf("wait() unexpected state = %+v, want Waiting", d.state)
 			}
@@ -83,7 +83,7 @@ func Test_delay_wait(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			d := &delay{
+			d := &delay[any]{
 				state:         tt.fields.state,
 				idleChannel:   tt.fields.idleChannel,
 				egressChannel: tt.fields.egressChannel,
@@ -111,7 +111,7 @@ func Test_delay_available(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			d := &delay{
+			d := &delay[any]{
 				state:         tt.fields.state,
 				egressChannel: tt.fields.egressChannel,
 				cancelChannel: tt.fields.cancel,

dispatcher.go

@@ -18,44 +18,44 @@ const (
 )
 
 // Dispatcher processes the ingress and dispatching of scheduled messages.
-type Dispatcher struct {
+type Dispatcher[T any] struct {
 	state       dispatcherState
 	maxMessages int
 
-	pq          priorityQueue
-	nextMessage *ScheduledMessage
-	delayer     delayer
+	pq          priorityQueue[T]
+	nextMessage *ScheduledMessage[T]
+	delayer     delayer[T]
 
 	delayerIdleChannel chan bool
-	dispatchChannel    chan interface{}
-	ingressChannel     chan *ScheduledMessage
+	dispatchChannel    chan T
+	ingressChannel     chan *ScheduledMessage[T]
 	shutdown           chan error
 	stopProcess        chan bool
 
 	mutex *sync.Mutex
 }
 
 // NewDispatcher creates a new instance of a Dispatcher.
-func NewDispatcher(config *DispatcherConfig) (*Dispatcher, error) {
+func NewDispatcher[T any](config *DispatcherConfig) (*Dispatcher[T], error) {
 	if config.MaxMessages <= 0 {
 		return nil, errors.New("MaxMessages should be greater than 0")
 	}
 
 	newIdleChannel := make(chan bool, 1)
-	newDispatchChannel := make(chan interface{}, config.DispatchChannelSize)
-	newPq := priorityQueue{
-		items:         make([]*item, 0),
+	newDispatchChannel := make(chan T, config.DispatchChannelSize)
+	newPq := priorityQueue[T]{
+		items:         make([]*item[T], 0),
 		maintainOrder: config.GuaranteeOrder,
 	}
 
 	heap.Init(&newPq)
-	return &Dispatcher{
+	return &Dispatcher[T]{
 		pq:                 newPq,
 		maxMessages:        config.MaxMessages,
-		delayer:            newDelay(newDispatchChannel, newIdleChannel),
+		delayer:            newDelay[T](newDispatchChannel, newIdleChannel),
 		delayerIdleChannel: newIdleChannel,
 		dispatchChannel:    newDispatchChannel,
-		ingressChannel:     make(chan *ScheduledMessage, config.IngressChannelSize),
+		ingressChannel:     make(chan *ScheduledMessage[T], config.IngressChannelSize),
 		shutdown:           make(chan error),
 		stopProcess:        make(chan bool),
 		mutex:              &sync.Mutex{},
@@ -67,7 +67,7 @@ func NewDispatcher(config *DispatcherConfig) (*Dispatcher, error) {
 //
 // If drainImmediately is true, then all messages will be dispatched immediately regardless of the schedule set. Order
 // can be lost if new messages are still being ingested.
-func (d *Dispatcher) Shutdown(ctx context.Context, drainImmediately bool) error {
+func (d *Dispatcher[T]) Shutdown(ctx context.Context, drainImmediately bool) error {
 	if d.state == shutdown || d.state == shutdownAndDrain {
 		return errors.New("shutdown has already happened")
 	}
@@ -109,7 +109,7 @@ func (d *Dispatcher) Shutdown(ctx context.Context, drainImmediately bool) error
 }
 
 // Start initializes the processing of scheduled messages and blocks.
-func (d *Dispatcher) Start() error {
+func (d *Dispatcher[T]) Start() error {
 	d.mutex.Lock()
 	if d.state == shutdown || d.state == shutdownAndDrain {
 		return errors.New("dispatcher is already running and shutting/shut down")
@@ -124,7 +124,7 @@ func (d *Dispatcher) Start() error {
 }
 
 // Pause updates the state of the Dispatcher to stop processing messages and will close the main process loop.
-func (d *Dispatcher) Pause() error {
+func (d *Dispatcher[T]) Pause() error {
 	d.mutex.Lock()
 	if d.state == shutdown || d.state == shutdownAndDrain {
 		return errors.New("dispatcher is shutting/shut down and cannot be paused")
@@ -141,7 +141,7 @@ func (d *Dispatcher) Pause() error {
 
 // Resume updates the state of the Dispatcher to start processing messages and starts the timer for the last message
 // being processed and blocks.
-func (d *Dispatcher) Resume() error {
+func (d *Dispatcher[T]) Resume() error {
 	d.mutex.Lock()
 	if d.state == shutdown || d.state == shutdownAndDrain {
 		return errors.New("dispatcher is shutting/shut down")
@@ -159,7 +159,7 @@ func (d *Dispatcher) Resume() error {
 }
 
 // process handles the processing of scheduled messages.
-func (d *Dispatcher) process() {
+func (d *Dispatcher[T]) process() {
 	for {
 		select {
 		case <-d.stopProcess:
@@ -180,7 +180,7 @@ func (d *Dispatcher) process() {
 }
 
 // handleShutdown drains the heap.
-func (d *Dispatcher) handleShutdownAndDrain() {
+func (d *Dispatcher[T]) handleShutdownAndDrain() {
 	if d.state == shutdownAndDrain {
 		d.delayer.stop(true)
 		if len(d.delayerIdleChannel) > 0 {
@@ -192,7 +192,7 @@ func (d *Dispatcher) handleShutdownAndDrain() {
 
 // handlePriorityQueue checks whether the heap is full and will Pop the next message if present and when the delayer is
 // idle.
-func (d *Dispatcher) handlePriorityQueue() (cont bool) {
+func (d *Dispatcher[T]) handlePriorityQueue() (cont bool) {
 	// check if we've exceeded the maximum messages to store in the heap
 	if d.pq.Len() >= d.maxMessages {
 		if len(d.delayerIdleChannel) > 0 {
@@ -210,7 +210,7 @@ func (d *Dispatcher) handlePriorityQueue() (cont bool) {
 
 // handleIngress checks for new messages off the ingress channel and will either dispatch if `shutdownAndDrain`, replace
 // the current delayer message or add to the heap.
-func (d *Dispatcher) handleIngress() {
+func (d *Dispatcher[T]) handleIngress() {
 	if len(d.ingressChannel) > 0 {
 		if msg, ok := <-d.ingressChannel; ok {
 			if d.state == shutdownAndDrain {
@@ -232,26 +232,26 @@ func (d *Dispatcher) handleIngress() {
 	}
 }
 
-func (d *Dispatcher) waitNextMessage() {
-	msg := heap.Pop(&d.pq).(*ScheduledMessage)
+func (d *Dispatcher[T]) waitNextMessage() {
+	msg := heap.Pop(&d.pq).(*ScheduledMessage[T])
 	d.nextMessage = msg
 	d.delayer.wait(msg)
 }
 
-func (d *Dispatcher) drainHeap() {
+func (d *Dispatcher[T]) drainHeap() {
 	for d.pq.Len() > 0 {
-		msg := heap.Pop(&d.pq).(*ScheduledMessage)
+		msg := heap.Pop(&d.pq).(*ScheduledMessage[T])
 		// dispatch the message immediately
 		d.dispatchChannel <- msg.Message
 	}
 }
 
 // IngressChannel returns the send-only channel of type `ScheduledMessage`.
-func (d *Dispatcher) IngressChannel() chan<- *ScheduledMessage {
+func (d *Dispatcher[T]) IngressChannel() chan<- *ScheduledMessage[T] {
 	return d.ingressChannel
 }
 
-// DispatchChannel returns a receive-only channel of type `interface{}`.
-func (d *Dispatcher) DispatchChannel() <-chan interface{} {
+// DispatchChannel returns a receive-only channel of type `T`.
+func (d *Dispatcher[T]) DispatchChannel() <-chan T {
 	return d.dispatchChannel
 }