-
Notifications
You must be signed in to change notification settings - Fork 61
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add custom fallback functions to the finality listener managers
Signed-off-by: Angelo De Caro <[email protected]>
- Loading branch information
Showing
6 changed files
with
191 additions
and
9 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,85 @@ | ||
| /* | ||
| Copyright IBM Corp. All Rights Reserved. | ||
| SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package finality | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "strings" | ||
|
|
||
| "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" | ||
| "github.com/hyperledger-labs/fabric-smart-client/platform/common/driver" | ||
| "github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/collections" | ||
| "github.com/hyperledger-labs/fabric-smart-client/platform/fabric" | ||
| "github.com/hyperledger-labs/fabric-smart-client/platform/fabric/core/generic/finality" | ||
| ) | ||
|
|
||
| type DeliveryScanQueryByID struct { | ||
| Delivery *fabric.Delivery | ||
| Ledger *fabric.Ledger | ||
| Mapper finality.TxInfoMapper[TxInfo] | ||
| } | ||
|
|
||
| func (q *DeliveryScanQueryByID) QueryByID(evicted map[driver.TxID][]finality.ListenerEntry[TxInfo]) (<-chan []TxInfo, error) { | ||
| txIDs := collections.Keys(evicted) | ||
| logger.Debugf("Launching routine to scan for txs [%v]", txIDs) | ||
|
|
||
| results := collections.NewSet(txIDs...) | ||
| ch := make(chan []TxInfo, len(txIDs)) | ||
|
|
||
| // for each txID, fetch the corresponding transaction. | ||
| // if the transaction is not found, start a delivery for it | ||
| for _, txID := range txIDs { | ||
| logger.Debugf("loading transaction [%s] from ledger...", txID) | ||
| pt, err := q.Ledger.GetTransactionByID(txID) | ||
| if err == nil { | ||
| logger.Debugf("transaction [%s] found on ledger", txID) | ||
| infos, err := q.Mapper.MapProcessedTx(pt) | ||
| if err != nil { | ||
| return nil, errors.Wrapf(err, "failed to map tx [%s]", txID) | ||
| } | ||
| ch <- infos | ||
| continue | ||
| } | ||
|
|
||
| // which kind of error do we have here? | ||
| if strings.Contains(err.Error(), fmt.Sprintf("TXID [%s] not available", txID)) { | ||
| // transaction was not found | ||
| logger.Errorf("tx [%s] not found on the ledger [%s], start a scan for future transactions", txID, err) | ||
| // start delivery for the future | ||
| // TODO: find a better starting point | ||
| err := q.Delivery.Scan(context.TODO(), "", func(tx *fabric.ProcessedTransaction) (bool, error) { | ||
| if !results.Contains(tx.TxID()) { | ||
| return false, nil | ||
| } | ||
|
|
||
| logger.Debugf("Received result for tx [%s, %v, %d]...", tx.TxID(), tx.ValidationCode(), len(tx.Results())) | ||
| infos, err := q.Mapper.MapProcessedTx(tx) | ||
| if err != nil { | ||
| logger.Errorf("failed mapping tx [%s]: %v", tx.TxID(), err) | ||
| return true, err | ||
| } | ||
| ch <- infos | ||
| results.Remove(tx.TxID()) | ||
|
|
||
| return results.Length() == 0, nil | ||
| }) | ||
| if err != nil { | ||
| logger.Errorf("Failed scanning: %v", err) | ||
| return nil, err | ||
| } | ||
|
|
||
| continue | ||
| } | ||
|
|
||
| // error not recoverable, fail | ||
| logger.Debugf("scan for tx [%s] failed with err [%s]", txID, err) | ||
| return nil, errors.Wrapf(err, "failed scanning tx [%s]", txID) | ||
| } | ||
|
|
||
| return ch, nil | ||
| } |
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,70 @@ | ||
| /* | ||
| Copyright IBM Corp. All Rights Reserved. | ||
| SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package lookup | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
|
|
||
| driver2 "github.com/hyperledger-labs/fabric-smart-client/platform/common/driver" | ||
| "github.com/hyperledger-labs/fabric-smart-client/platform/fabric" | ||
| "github.com/hyperledger-labs/fabric-smart-client/platform/fabric/core/generic/finality" | ||
| "github.com/hyperledger-labs/fabric-token-sdk/token/services/network/fabric/tcc" | ||
| "github.com/pkg/errors" | ||
| ) | ||
|
|
||
| const ( | ||
| QueryStates = tcc.QueryStates | ||
| ) | ||
|
|
||
| type DeliveryScanQueryByID struct { | ||
| Channel *fabric.Channel | ||
| } | ||
|
|
||
| func (q *DeliveryScanQueryByID) QueryByID(evicted map[driver2.TxID][]finality.ListenerEntry[TxInfo]) (<-chan []TxInfo, error) { | ||
| // collects keys by namespace | ||
| keysByNS := map[driver2.Namespace][]string{} | ||
| for k, v := range evicted { | ||
| ns := v[0].Namespace() | ||
| _, ok := keysByNS[ns] | ||
| if !ok { | ||
| keysByNS[ns] = []string{} | ||
| } | ||
| keysByNS[ns] = append(keysByNS[ns], k) | ||
| } | ||
|
|
||
| ch := make(chan []TxInfo, len(evicted)) | ||
| // for each namespace, have a call to the token chaincode | ||
| for ns, keys := range keysByNS { | ||
| arg, err := json.Marshal(keys) | ||
| if err != nil { | ||
| return nil, errors.Wrapf(err, "failed marshalling args for query by ids [%v]", keys) | ||
| } | ||
|
|
||
| logger.Debugf("querying chaincode [%s] for the states of ids [%v]", ns, keys) | ||
| chaincode := q.Channel.Chaincode(ns) | ||
| res, err := chaincode.Query(QueryStates, arg).Query() | ||
| if err != nil { | ||
| return nil, errors.Wrapf(err, "failed querying by ids [%v]", keys) | ||
| } | ||
| values := make([][]byte, 0, len(keys)) | ||
| err = json.Unmarshal(res, &values) | ||
| if err != nil { | ||
| return nil, errors.Wrapf(err, "failed unmarshalling results for query by ids [%v]", keys) | ||
| } | ||
| infos := make([]TxInfo, 0, len(values)) | ||
| for i, value := range values { | ||
| infos = append(infos, TxInfo{ | ||
| Namespace: ns, | ||
| Key: keys[i], | ||
| Value: value, | ||
| }) | ||
| } | ||
| ch <- infos | ||
| } | ||
|
|
||
| return ch, nil | ||
| } |