The highest tagged major version is
README
¶
Notificationcenter stream library
License Apache 2.0
The union eventstream wrapper over nifferent stream implementations.
Using examples
Create new stream processor
// Create new stream processor
eventStream, err = nats.NewStream([]string{"event"}, "nats://hostname:4222")
if err != nil {
log.Fatal(err)
}
// Register stream processor
err = notificationcenter.Register("events", eventStream)
if err != nil {
log.Fatal(err)
}
Send event to the notification stream
// Send by global functions
notificationcenter.Send("events", message{title: "event 1"})
// Send by logger interface
events := notificationcenter.StreamByName("events")
events.Send(message{title: "event 2"})
Subscribe for the specific notification stream
import (
nc "github.com/geniusrabbit/notificationcenter"
"github.com/geniusrabbit/notificationcenter/nats"
)
func main() {
events := nats.MustNewSubscriber("nats://connection", "group", []string{"events"})
nc.Register("events", events)
// Add new handler to process the stream "events"
nc.Subscribe("events", notificationcenter.FuncHandler(func(msg nc.Message) error {
fmt.Printf("%v\n", msg.Data())
return nil
}))
// Run seubscribers listeners
nc.Listen()
}
TODO
- remove metrics from the stream (DEPRECATED)
- Add support NATS & NATS stream
- Add kafka support
Documentation
¶
Index ¶
- Variables
- func Close()
- func Listen() (err error)
- func OnClose() <-chan bool
- func Register(name string, elem ...interface{}) error
- func Send(name string, msg ...interface{}) error
- func Subscribe(name string, h Handler) error
- func Unregister(elem ...interface{}) error
- func UnregisterAllByName(name string, _streamers, _subscribers bool)
- func Unsubscribe(name string, h Handler) error
- type FuncHandler
- type Handler
- type Message
- type MultithreadHandler
- type Streamer
- type Subscriber
Constants ¶
This section is empty.
Variables ¶
var ( ErrInvalidObject = errors.New("[notificationcenter] invalid handler") ErrInterfaceAlreadySubscribed = errors.New("[notificationcenter] interface already subscribed") ErrInvalidParams = errors.New("[notificationcenter] invalid params") )
Errors set
Functions ¶
func OnClose ¶
func OnClose() <-chan bool
OnClose event will be executed only after closing all interfaces
Usecases in the application makes subsribing for the finishing event very convinient ```go
func myDatabaseObserver() {
<- notificationcenter.OnClose()
// ... Do something
}
```
func Register ¶
Register the new streamer or subscriber interface to future using Object can implement both interface and process all events from start to end
func Unregister ¶
func Unregister(elem ...interface{}) error
Unregister exist streamer or subscriber interface from the global storage
func UnregisterAllByName ¶
UnregisterAllByName exist streamer or subscriber interface from the global storage
func Unsubscribe ¶
Unsubscribe some paticular handler interface from subscriber with the *name*
Types ¶
type FuncHandler ¶
FuncHandler type
type Message ¶
type Message interface {
// Unical message ID (depends on transport)
ID() string
// Body returns message data as bytes
Body() []byte
// Acknowledgment of the message processing
Ack() error
}
Message describes the access methods to the message original object
type MultithreadHandler ¶
MultithreadHandler it's ext interface which contains count of concurrent processes
func NewMultithreadHandler ¶
func NewMultithreadHandler(count int, handler Handler) MultithreadHandler
NewMultithreadHandler processor
type Streamer ¶
type Streamer interface {
// Send data to statistic
Send(messages ...interface{}) error
}
Streamer pipeline base declaration
func StreamByName ¶
StreamByName returns streamer interface by codename if exists
Source Files
Directories