lotus/paychmgr/msglistener.go
2022-06-14 17:00:51 +02:00

55 lines
1.2 KiB
Go

package paychmgr
import (
"github.com/hannahhoward/go-pubsub"
"github.com/ipfs/go-cid"
"golang.org/x/xerrors"
)
type msgListeners struct {
ps *pubsub.PubSub
}
type msgCompleteEvt struct {
mcid cid.Cid
err error
}
type subscriberFn func(msgCompleteEvt)
func newMsgListeners() msgListeners {
ps := pubsub.New(func(event pubsub.Event, subFn pubsub.SubscriberFn) error {
evt, ok := event.(msgCompleteEvt)
if !ok {
return xerrors.Errorf("wrong type of event")
}
sub, ok := subFn.(subscriberFn)
if !ok {
return xerrors.Errorf("wrong type of subscriber")
}
sub(evt)
return nil
})
return msgListeners{ps: ps}
}
// onMsgComplete registers a callback for when the message with the given cid
// completes
func (ml *msgListeners) onMsgComplete(mcid cid.Cid, cb func(error)) pubsub.Unsubscribe {
var fn subscriberFn = func(evt msgCompleteEvt) {
if mcid.Equals(evt.mcid) {
cb(evt.err)
}
}
return ml.ps.Subscribe(fn)
}
// fireMsgComplete is called when a message completes
func (ml *msgListeners) fireMsgComplete(mcid cid.Cid, err error) {
e := ml.ps.Publish(msgCompleteEvt{mcid: mcid, err: err})
if e != nil {
// In theory we shouldn't ever get an error here
log.Errorf("unexpected error publishing message complete: %s", e)
}
}