mirror of
https://github.com/SecurityBrewery/catalyst.git
synced 2026-02-20 12:05:27 +01:00
Jonas Plum
GitHub
105
bus/bus.go
105
bus/bus.go
@@ -1,70 +1,69 @@
|
||||
package bus
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log"
|
||||
"github.com/arangodb/go-driver"
|
||||
|
||||
emitter "github.com/emitter-io/go/v2"
|
||||
"github.com/SecurityBrewery/catalyst/generated/model"
|
||||
)
|
||||
|
||||
type ResultMsg struct {
|
||||
Automation string `json:"automation"`
|
||||
Data map[string]any `json:"data,omitempty"`
|
||||
Target *model.Origin `json:"target"`
|
||||
}
|
||||
|
||||
type RequestMsg struct {
|
||||
IDs []driver.DocumentID `json:"ids"`
|
||||
Function string `json:"function"`
|
||||
User string `json:"user"`
|
||||
}
|
||||
|
||||
type JobMsg struct {
|
||||
ID string `json:"id"`
|
||||
Automation string `json:"automation"`
|
||||
Origin *model.Origin `json:"origin"`
|
||||
Message *model.Message `json:"message"`
|
||||
}
|
||||
|
||||
type DatabaseUpdateType string
|
||||
|
||||
const (
|
||||
DatabaseEntryRead DatabaseUpdateType = "read"
|
||||
DatabaseEntryCreated DatabaseUpdateType = "created"
|
||||
DatabaseEntryUpdated DatabaseUpdateType = "updated"
|
||||
)
|
||||
|
||||
type DatabaseUpdateMsg struct {
|
||||
IDs []driver.DocumentID `json:"ids"`
|
||||
Type DatabaseUpdateType `json:"type"`
|
||||
}
|
||||
|
||||
type Bus struct {
|
||||
config *Config
|
||||
client *emitter.Client
|
||||
ResultChannel *Channel[*ResultMsg]
|
||||
RequestChannel *Channel[*RequestMsg]
|
||||
JobChannel *Channel[*JobMsg]
|
||||
DatabaseChannel *Channel[*DatabaseUpdateMsg]
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
Host string
|
||||
Key string
|
||||
databaseUpdateBusKey string
|
||||
jobBusKey string
|
||||
resultBusKey string
|
||||
requestKey string
|
||||
APIUrl string
|
||||
func New() *Bus {
|
||||
return &Bus{
|
||||
ResultChannel: &Channel[*ResultMsg]{},
|
||||
RequestChannel: &Channel[*RequestMsg]{},
|
||||
JobChannel: &Channel[*JobMsg]{},
|
||||
DatabaseChannel: &Channel[*DatabaseUpdateMsg]{},
|
||||
}
|
||||
}
|
||||
|
||||
func New(c *Config) (*Bus, error) {
|
||||
client, err := emitter.Connect(c.Host, func(_ *emitter.Client, msg emitter.Message) {
|
||||
log.Printf("received: '%s' topic: '%s'\n", msg.Payload(), msg.Topic())
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
c.databaseUpdateBusKey, err = client.GenerateKey(c.Key, channelDatabaseUpdate+"/", "rwls", 0)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.jobBusKey, err = client.GenerateKey(c.Key, channelJob+"/", "rwls", 0)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.resultBusKey, err = client.GenerateKey(c.Key, channelResult+"/", "rwls", 0)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.requestKey, err = client.GenerateKey(c.Key, ChannelRequest+"/", "rwls", 0)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &Bus{config: c, client: client}, err
|
||||
type Channel[T any] struct {
|
||||
Subscriber []func(T)
|
||||
}
|
||||
|
||||
func (b *Bus) jsonPublish(msg any, channel, key string) error {
|
||||
payload, err := json.Marshal(msg)
|
||||
if err != nil {
|
||||
return err
|
||||
func (c *Channel[T]) Publish(msg T) {
|
||||
for _, s := range c.Subscriber {
|
||||
go s(msg)
|
||||
}
|
||||
|
||||
return b.client.Publish(key, channel, payload)
|
||||
}
|
||||
|
||||
func (b *Bus) safeSubscribe(key, channel string, handler func(c *emitter.Client, m emitter.Message)) error {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
log.Printf("Recovered %s in channel %s\n", r, channel)
|
||||
}
|
||||
}()
|
||||
|
||||
return b.client.Subscribe(key, channel, handler)
|
||||
func (c *Channel[T]) Subscribe(handler func(T)) {
|
||||
c.Subscriber = append(c.Subscriber, handler)
|
||||
}
|
||||
|
||||
@@ -1,43 +0,0 @@
|
||||
package bus
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log"
|
||||
|
||||
"github.com/arangodb/go-driver"
|
||||
emitter "github.com/emitter-io/go/v2"
|
||||
)
|
||||
|
||||
const channelDatabaseUpdate = "databaseupdate"
|
||||
|
||||
type DatabaseUpdateType string
|
||||
|
||||
const (
|
||||
DatabaseEntryRead DatabaseUpdateType = "read"
|
||||
DatabaseEntryCreated DatabaseUpdateType = "created"
|
||||
DatabaseEntryUpdated DatabaseUpdateType = "updated"
|
||||
)
|
||||
|
||||
type DatabaseUpdateMsg struct {
|
||||
IDs []driver.DocumentID `json:"ids"`
|
||||
Type DatabaseUpdateType `json:"type"`
|
||||
}
|
||||
|
||||
func (b *Bus) PublishDatabaseUpdate(ids []driver.DocumentID, databaseUpdateType DatabaseUpdateType) error {
|
||||
return b.jsonPublish(&DatabaseUpdateMsg{
|
||||
IDs: ids,
|
||||
Type: databaseUpdateType,
|
||||
}, channelDatabaseUpdate, b.config.databaseUpdateBusKey)
|
||||
}
|
||||
|
||||
func (b *Bus) SubscribeDatabaseUpdate(f func(msg *DatabaseUpdateMsg)) error {
|
||||
return b.safeSubscribe(b.config.databaseUpdateBusKey, channelDatabaseUpdate, func(c *emitter.Client, m emitter.Message) {
|
||||
var msg DatabaseUpdateMsg
|
||||
if err := json.Unmarshal(m.Payload(), &msg); err != nil {
|
||||
log.Println(err)
|
||||
|
||||
return
|
||||
}
|
||||
go f(&msg)
|
||||
})
|
||||
}
|
||||
43
bus/job.go
43
bus/job.go
@@ -1,43 +0,0 @@
|
||||
package bus
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log"
|
||||
|
||||
emitter "github.com/emitter-io/go/v2"
|
||||
|
||||
"github.com/SecurityBrewery/catalyst/generated/model"
|
||||
)
|
||||
|
||||
const channelJob = "job"
|
||||
|
||||
type JobMsg struct {
|
||||
ID string `json:"id"`
|
||||
Automation string `json:"automation"`
|
||||
Origin *model.Origin `json:"origin"`
|
||||
Message *model.Message `json:"message"`
|
||||
}
|
||||
|
||||
func (b *Bus) PublishJob(id, automation string, payload any, context *model.Context, origin *model.Origin) error {
|
||||
return b.jsonPublish(&JobMsg{
|
||||
ID: id,
|
||||
Automation: automation,
|
||||
Origin: origin,
|
||||
Message: &model.Message{
|
||||
Context: context,
|
||||
Payload: payload,
|
||||
},
|
||||
}, channelJob, b.config.jobBusKey)
|
||||
}
|
||||
|
||||
func (b *Bus) SubscribeJob(f func(msg *JobMsg)) error {
|
||||
return b.safeSubscribe(b.config.jobBusKey, channelJob, func(c *emitter.Client, m emitter.Message) {
|
||||
var msg JobMsg
|
||||
if err := json.Unmarshal(m.Payload(), &msg); err != nil {
|
||||
log.Println(err)
|
||||
|
||||
return
|
||||
}
|
||||
go f(&msg)
|
||||
})
|
||||
}
|
||||
@@ -1,37 +0,0 @@
|
||||
package bus
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log"
|
||||
|
||||
"github.com/arangodb/go-driver"
|
||||
emitter "github.com/emitter-io/go/v2"
|
||||
)
|
||||
|
||||
const ChannelRequest = "request"
|
||||
|
||||
type RequestMsg struct {
|
||||
IDs []driver.DocumentID `json:"ids"`
|
||||
Function string `json:"function"`
|
||||
User string `json:"user"`
|
||||
}
|
||||
|
||||
func (b *Bus) PublishRequest(user, f string, ids []driver.DocumentID) error {
|
||||
return b.jsonPublish(&RequestMsg{
|
||||
User: user,
|
||||
Function: f,
|
||||
IDs: ids,
|
||||
}, ChannelRequest, b.config.requestKey)
|
||||
}
|
||||
|
||||
func (b *Bus) SubscribeRequest(f func(msg *RequestMsg)) error {
|
||||
return b.safeSubscribe(b.config.requestKey, ChannelRequest, func(c *emitter.Client, m emitter.Message) {
|
||||
msg := &RequestMsg{}
|
||||
if err := json.Unmarshal(m.Payload(), msg); err != nil {
|
||||
log.Println(err)
|
||||
|
||||
return
|
||||
}
|
||||
go f(msg)
|
||||
})
|
||||
}
|
||||
@@ -1,34 +0,0 @@
|
||||
package bus
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log"
|
||||
|
||||
emitter "github.com/emitter-io/go/v2"
|
||||
|
||||
"github.com/SecurityBrewery/catalyst/generated/model"
|
||||
)
|
||||
|
||||
const channelResult = "result"
|
||||
|
||||
type ResultMsg struct {
|
||||
Automation string `json:"automation"`
|
||||
Data map[string]any `json:"data,omitempty"`
|
||||
Target *model.Origin `json:"target"`
|
||||
}
|
||||
|
||||
func (b *Bus) PublishResult(automation string, data map[string]any, target *model.Origin) error {
|
||||
return b.jsonPublish(&ResultMsg{Automation: automation, Data: data, Target: target}, channelResult, b.config.resultBusKey)
|
||||
}
|
||||
|
||||
func (b *Bus) SubscribeResult(f func(msg *ResultMsg)) error {
|
||||
return b.safeSubscribe(b.config.resultBusKey, channelResult, func(c *emitter.Client, m emitter.Message) {
|
||||
msg := &ResultMsg{}
|
||||
if err := json.Unmarshal(m.Payload(), msg); err != nil {
|
||||
log.Println(err)
|
||||
|
||||
return
|
||||
}
|
||||
go f(msg)
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user