message-bus - Go Packages (original) (raw)

Package messagebus provides simple async message publisher

package main

import ( "fmt" "sync"

messagebus "github.com/vardius/message-bus"

)

func main() { queueSize := 100 bus := messagebus.New(queueSize)

var wg sync.WaitGroup
wg.Add(2)

_ = bus.Subscribe("topic", func(v bool) {
    defer wg.Done()
    fmt.Println("s1", v)
})

_ = bus.Subscribe("topic", func(v bool) {
    defer wg.Done()
    fmt.Println("s2", v)
})

// Publish block only when the buffer of one of the subscribers is full.
// change the buffer size altering queueSize when creating new messagebus
bus.Publish("topic", true)
wg.Wait()

}

Output:

s1 true s2 true

package main

import ( "fmt"

messagebus "github.com/vardius/message-bus"

)

func main() { queueSize := 2 subscribersAmount := 3

ch := make(chan int, queueSize)
defer close(ch)

bus := messagebus.New(queueSize)

for i := 0; i < subscribersAmount; i++ {
    _ = bus.Subscribe("topic", func(i int, out chan<- int) { out <- i })
}

go func() {
    for n := 0; n < queueSize; n++ {
        bus.Publish("topic", n, ch)
    }
}()

var sum = 0
for sum < (subscribersAmount * queueSize) {
    select {
    case <-ch:
        sum++
    }
}

fmt.Println(sum)

}

Output:

type MessageBus
- func New(handlerQueueSize int) MessageBus
Package
Package (Second)

This section is empty.

type MessageBus interface {

Publish(topic [string](/builtin#string), args ...interface{})

Close(topic [string](/builtin#string))

Subscribe(topic [string](/builtin#string), fn interface{}) [error](/builtin#error)

Unsubscribe(topic [string](/builtin#string), fn interface{}) [error](/builtin#error)

}

MessageBus implements publish/subscribe messaging paradigm

func New(handlerQueueSize int) MessageBus

New creates new MessageBus handlerQueueSize sets buffered channel length per subscriber