-
Notifications
You must be signed in to change notification settings - Fork 8.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
FABGW-8: Wait for transaction commits
Initial implementation of commit waiting capability for the embedded Gateway. Not yet integrated into Gateway services. Signed-off-by: Mark S. Lewis <mark_lewis@uk.ibm.com>
- Loading branch information
1 parent
b40f703
commit 76763d1
Showing
10 changed files
with
689 additions
and
16 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,148 @@ | ||
| /* | ||
| Copyright 2021 IBM All Rights Reserved. | ||
| SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package commit | ||
|
|
||
| import ( | ||
| "sync" | ||
|
|
||
| "github.com/hyperledger/fabric/core/ledger" | ||
| ) | ||
|
|
||
| type channelLevelNotifier struct { | ||
| commitChannel <-chan *ledger.CommitNotification | ||
| lock sync.Mutex | ||
| listeners map[string][]*transactionListener | ||
| done <-chan struct{} | ||
| } | ||
|
|
||
| func newChannelNotifier(done <-chan struct{}, commitChannel <-chan *ledger.CommitNotification) *channelLevelNotifier { | ||
| notifier := &channelLevelNotifier{ | ||
| commitChannel: commitChannel, | ||
| listeners: make(map[string][]*transactionListener), | ||
| done: done, | ||
| } | ||
| go notifier.run() | ||
| return notifier | ||
| } | ||
|
|
||
| func (notifier *channelLevelNotifier) run() { | ||
| for { | ||
| select { | ||
| case blockCommit, ok := <-notifier.commitChannel: | ||
| if !ok { | ||
| // This should never happen and behaviour is currently intentionally undefined / untested | ||
| notifier.close() | ||
| return | ||
| } | ||
| notifier.removeCompletedListeners() | ||
| notifier.receiveBlock(blockCommit) | ||
| case <-notifier.done: | ||
| notifier.close() | ||
| return | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (notifier *channelLevelNotifier) receiveBlock(blockCommit *ledger.CommitNotification) { | ||
| for transactionID, status := range blockCommit.TxIDValidationCodes { | ||
| notification := &Notification{ | ||
| BlockNumber: blockCommit.BlockNumber, | ||
| TransactionID: transactionID, | ||
| ValidationCode: status, | ||
| } | ||
| notifier.notify(notification) | ||
| } | ||
| } | ||
|
|
||
| func (notifier *channelLevelNotifier) removeCompletedListeners() { | ||
| notifier.lock.Lock() | ||
| defer notifier.lock.Unlock() | ||
|
|
||
| for key, listeners := range notifier.listeners { | ||
| for i := 0; i < len(listeners); { | ||
| if !listeners[i].isDone() { | ||
| i++ | ||
| continue | ||
| } | ||
|
|
||
| listeners[i].close() | ||
|
|
||
| lastIndex := len(listeners) - 1 | ||
| listeners[i] = listeners[lastIndex] | ||
| listeners = listeners[:lastIndex] | ||
| } | ||
|
|
||
| if len(listeners) > 0 { | ||
| notifier.listeners[key] = listeners | ||
| } else { | ||
| delete(notifier.listeners, key) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (notifier *channelLevelNotifier) notify(notification *Notification) { | ||
| notifier.lock.Lock() | ||
| defer notifier.lock.Unlock() | ||
|
|
||
| for _, listener := range notifier.listeners[notification.TransactionID] { | ||
| listener.receive(notification) | ||
| listener.close() | ||
| } | ||
|
|
||
| delete(notifier.listeners, notification.TransactionID) | ||
| } | ||
|
|
||
| func (notifier *channelLevelNotifier) registerListener(done <-chan struct{}, transactionID string) <-chan Notification { | ||
| notifyChannel := make(chan Notification, 1) // avoid blocking and only expect one notification per channel | ||
| listener := &transactionListener{ | ||
| done: done, | ||
| transactionID: transactionID, | ||
| notifyChannel: notifyChannel, | ||
| } | ||
|
|
||
| notifier.lock.Lock() | ||
| notifier.listeners[transactionID] = append(notifier.listeners[transactionID], listener) | ||
| notifier.lock.Unlock() | ||
|
|
||
| return notifyChannel | ||
| } | ||
|
|
||
| func (notifier *channelLevelNotifier) close() { | ||
| notifier.lock.Lock() | ||
| defer notifier.lock.Unlock() | ||
|
|
||
| for _, listeners := range notifier.listeners { | ||
| for _, listener := range listeners { | ||
| listener.close() | ||
| } | ||
| } | ||
|
|
||
| notifier.listeners = nil | ||
| } | ||
|
|
||
| type transactionListener struct { | ||
| done <-chan struct{} | ||
| transactionID string | ||
| notifyChannel chan<- Notification | ||
| } | ||
|
|
||
| func (listener *transactionListener) isDone() bool { | ||
| select { | ||
| case <-listener.done: | ||
| return true | ||
| default: | ||
| return false | ||
| } | ||
| } | ||
|
|
||
| func (listener *transactionListener) close() { | ||
| close(listener.notifyChannel) | ||
| } | ||
|
|
||
| func (listener *transactionListener) receive(notification *Notification) { | ||
| listener.notifyChannel <- *notification | ||
| } |
119 changes: 119 additions & 0 deletions
119
internal/pkg/gateway/commit/mock/notificationsupplier.go
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
Oops, something went wrong.