mirrors/ctrl
mirrors/ctrl
1
0
Fork 0
mirror of https://github.com/postmannen/ctrl.git synced 2024-12-14 12:37:31 +00:00
Code Activity

defering subReply.Unsubscribe

Browse source
This commit is contained in:
postmannen 2022-06-18 08:12:14 +02:00
parent 988bd04d7f
commit 6c900296c9
1 changed files with 23 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

42
process.go
View file

@ -283,17 +283,6 @@ func (p process) messageDeliverNats(natsMsgPayload []byte, natsMsgHeader nats.He
//
// Create a subscriber for the ACK reply message.
subReply, err := natsConn.SubscribeSync(msg.Reply)
if err != nil {
er := fmt.Errorf("error: nats SubscribeSync failed: failed to create reply message for subject: %v, error: %v", msg.Reply, err)
// sendErrorLogMessage(p.toRingbufferCh, node(p.node), er)
log.Printf("%v, waiting %ds before retrying\n", er, subscribeSyncTimer)
time.Sleep(time.Second * subscribeSyncTimer)
subReply.Unsubscribe()
retryAttempts++
return ErrACKSubscribeRetry
}
defer func() {
err := subReply.Unsubscribe()
@ -302,6 +291,18 @@ func (p process) messageDeliverNats(natsMsgPayload []byte, natsMsgHeader nats.He
}
}()
if err != nil {
er := fmt.Errorf("error: nats SubscribeSync failed: failed to create reply message for subject: %v, error: %v", msg.Reply, err)
// sendErrorLogMessage(p.toRingbufferCh, node(p.node), er)
log.Printf("%v, waiting %ds before retrying\n", er, subscribeSyncTimer)
time.Sleep(time.Second * subscribeSyncTimer)
// subReply.Unsubscribe()
retryAttempts++
return ErrACKSubscribeRetry
}
// Publish message
err = natsConn.PublishMsg(msg)
if err != nil {
@ -797,14 +798,17 @@ func (p process) publishAMessage(m Message, zEnc *zstd.Encoder, once sync.Once,
case "g": // gzip
var buf bytes.Buffer
gzipW := gzip.NewWriter(&buf)
_, err := gzipW.Write(natsMsgPayloadSerialized)
if err != nil {
log.Printf("error: failed to write gzip: %v\n", err)
gzipW.Close()
return
}
gzipW.Close()
func() {
gzipW := gzip.NewWriter(&buf)
defer gzipW.Close()
defer gzipW.Flush()
_, err := gzipW.Write(natsMsgPayloadSerialized)
if err != nil {
log.Printf("error: failed to write gzip: %v\n", err)
return
}
}()
natsMsgPayloadCompressed = buf.Bytes()
natsMsgHeader["cmp"] = []string{p.configuration.Compression}