eosswedenorg/thalos
1
0
Fork 0
mirror of https://github.com/eosswedenorg/thalos synced 2026-06-16 04:24:56 +02:00
Code Releases Activity

api/client.go: add handler for table delta

Browse source
This commit is contained in:
Henrik Hautakoski 2024-02-04 11:32:28 +01:00
parent 9143c1e06b
commit e7ad37c38a
1 changed files with 18 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

21
api/client.go
View file

@ -17,9 +17,10 @@ type Client struct {
// waitgroup for worker threads.
wg sync.WaitGroup
OnError func(error)
OnAction func(message.ActionTrace)
OnHeartbeat func(message.HeartBeat)
OnError func(error)
OnAction func(message.ActionTrace)
OnHeartbeat func(message.HeartBeat)
OnTableDelta func(message.TableDelta)
}
func NewClient(reader Reader, decoder message.Decoder) *Client {
@ -55,6 +56,18 @@ func (c *Client) actHandler(payload []byte) {
c.OnAction(act)
}
// TableDelta handler
func (c *Client) tableDeltaHandler(payload []byte) {
td := message.TableDelta{}
if err := c.decoder(payload, &td); err != nil {
if c.OnError != nil {
c.OnError(err)
}
return
}
c.OnTableDelta(td)
}
// HeartBeat handler
func (c *Client) hbHandler(payload []byte) {
var hb message.HeartBeat
@ -75,6 +88,8 @@ func (c *Client) Subscribe(channel Channel) error {
handler = c.hbHandler
case ActionChannel{}.Channel().Type():
handler = c.actHandler
case TableDeltaChannel{}.Channel().Type():
handler = c.tableDeltaHandler
default:
return fmt.Errorf("invalid channel type. %s", t)
}