aarifullin/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Issues Pull requests Projects Releases Packages Wiki Activity

[#770] pkg/morph/subscriber: Add subscription for notary

Browse source 
Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
This commit is contained in:
Pavel Karpy 2021-08-09 18:27:04 +03:00 committed by Pavel Karpy
parent e24489d33c
commit f0cbf2e99d
1 changed files with 35 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
pkg/morph/subscriber/subscriber.go
View file

@ -20,8 +20,9 @@ type (
Subscriber interface { Subscriber interface {
SubscribeForNotification(...util.Uint160) (<-chan *state.NotificationEvent, error) SubscribeForNotification(...util.Uint160) (<-chan *state.NotificationEvent, error)
UnsubscribeForNotification() UnsubscribeForNotification()
Close()
BlockNotifications() (<-chan *block.Block, error) BlockNotifications() (<-chan *block.Block, error)
SubscribeForNotaryRequests(mainTXSigner util.Uint160) (<-chan *response.NotaryRequestEvent, error)
Close()
} }
subscriber struct { subscriber struct {
@ -29,10 +30,12 @@ type (
log *zap.Logger log *zap.Logger
client *client.WSClient client *client.WSClient
notify chan *state.NotificationEvent notifyChan chan *state.NotificationEvent
notifyIDs map[util.Uint160]string notifyIDs map[util.Uint160]string
blockChan chan *block.Block blockChan chan *block.Block
notaryChan chan *response.NotaryRequestEvent
} }
// Params is a group of Subscriber constructor parameters. // Params is a group of Subscriber constructor parameters.
@ -81,7 +84,7 @@ func (s *subscriber) SubscribeForNotification(contracts ...util.Uint160) (<-chan
s.notifyIDs[contract] = id s.notifyIDs[contract] = id
} }
return s.notify, nil return s.notifyChan, nil
} }
func (s *subscriber) UnsubscribeForNotification() { func (s *subscriber) UnsubscribeForNotification() {
@ -112,6 +115,14 @@ func (s *subscriber) BlockNotifications() (<-chan *block.Block, error) {
return s.blockChan, nil return s.blockChan, nil
} }
func (s *subscriber) SubscribeForNotaryRequests(mainTXSigner util.Uint160) (<-chan *response.NotaryRequestEvent, error) {
if _, err := s.client.SubscribeForNotaryRequests(nil, &mainTXSigner); err != nil {
return nil, fmt.Errorf("could not subscribe for notary request events: %w", err)
}
return s.notaryChan, nil
}
func (s *subscriber) routeNotifications(ctx context.Context) { func (s *subscriber) routeNotifications(ctx context.Context) {
for { for {
select { select {
@ -119,9 +130,10 @@ func (s *subscriber) routeNotifications(ctx context.Context) {
return return
case notification, ok := <-s.client.Notifications: case notification, ok := <-s.client.Notifications:
if !ok { if !ok {
s.log.Warn("remote channel has been closed") s.log.Warn("remote notification channel has been closed")
close(s.notify) close(s.notifyChan)
close(s.blockChan) close(s.blockChan)
close(s.notaryChan)
return return
} }
@ -130,11 +142,11 @@ func (s *subscriber) routeNotifications(ctx context.Context) {
case response.NotificationEventID: case response.NotificationEventID:
notification, ok := notification.Value.(*state.NotificationEvent) notification, ok := notification.Value.(*state.NotificationEvent)
if !ok { if !ok {
s.log.Error("can't cast notify event to the notify struct") s.log.Error("can't cast notify event value to the notify struct")
continue continue
} }
s.notify <- notification s.notifyChan <- notification
case response.BlockEventID: case response.BlockEventID:
b, ok := notification.Value.(*block.Block) b, ok := notification.Value.(*block.Block)
if !ok { if !ok {
@ -143,6 +155,14 @@ func (s *subscriber) routeNotifications(ctx context.Context) {
} }
s.blockChan <- b s.blockChan <- b
case response.NotaryRequestEventID:
notaryRequest, ok := notification.Value.(*response.NotaryRequestEvent)
if !ok {
s.log.Error("can't cast notify event value to the notary request struct")
continue
}
s.notaryChan <- notaryRequest
default: default:
s.log.Debug("unsupported notification from the chain", s.log.Debug("unsupported notification from the chain",
zap.Uint8("type", uint8(notification.Type)), zap.Uint8("type", uint8(notification.Type)),
@ -173,12 +193,13 @@ func New(ctx context.Context, p *Params) (Subscriber, error) {
} }
sub := &subscriber{ sub := &subscriber{
RWMutex: new(sync.RWMutex), RWMutex: new(sync.RWMutex),
log: p.Log, log: p.Log,
client: wsClient, client: wsClient,
notify: make(chan *state.NotificationEvent), notifyChan: make(chan *state.NotificationEvent),
notifyIDs: make(map[util.Uint160]string), notifyIDs: make(map[util.Uint160]string),
blockChan: make(chan *block.Block), blockChan: make(chan *block.Block),
notaryChan: make(chan *response.NotaryRequestEvent),
} }
// Worker listens all events from neo-go websocket and puts them // Worker listens all events from neo-go websocket and puts them