eosswedenorg/thalos
1
0
Fork 0
mirror of https://github.com/eosswedenorg/thalos synced 2026-07-03 11:53:41 +02:00
Code Releases Activity

ship_processor.go: Use publisher instead of redis directly.

Browse source
This commit is contained in:
Henrik Hautakoski 2023-01-13 13:33:01 +01:00
parent eec834a338
commit d0782c3c7c
1 changed files with 5 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

10
ship_processor.go
View file

@ -36,7 +36,7 @@ func encodeMessage(v interface{}) ([]byte, bool) {
func queueMessage(channel redis.ChannelInterface, payload []byte) bool { func queueMessage(channel redis.ChannelInterface, payload []byte) bool {
key := redisNs.NewKey(channel) key := redisNs.NewKey(channel)
err := redis.RegisterPublish(key.String(), payload).Err() err := publisher.Publish(key.String(), payload)
if err != nil { if err != nil {
log.WithError(err).Errorf("Failed to post to channel '%s'", key) log.WithError(err).Errorf("Failed to post to channel '%s'", key)
return false return false
@ -67,9 +67,9 @@ func processBlock(block *ship.GetBlocksResultV0) {
encodeQueue(redis.HeartbeatChannel, hb) encodeQueue(redis.HeartbeatChannel, hb)
_, err := redis.Send() err := publisher.Flush()
if err != nil { if err != nil {
log.WithError(err).Error("Failed to send redis") log.WithError(err).Error("Failed to send messages")
} }
} }
} }
@ -120,8 +120,8 @@ func processTraces(traces []*ship.TransactionTraceV0) {
} }
} }
_, err := redis.Send() err := publisher.Flush()
if err != nil { if err != nil {
log.WithError(err).Error("Failed to send redis") log.WithError(err).Error("Failed to send messages")
} }
} }