57 lines
1.2 KiB
Go
57 lines
1.2 KiB
Go
package paychmgr
|
|
|
|
import (
|
|
"golang.org/x/xerrors"
|
|
|
|
"github.com/hannahhoward/go-pubsub"
|
|
|
|
"github.com/ipfs/go-cid"
|
|
)
|
|
|
|
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)
|
|
}
|
|
}
|