Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Refactor of ChaincodeEvents service implementation to support resume #3283

Merged
merged 1 commit into from
Mar 15, 2022
Merged

Refactor of ChaincodeEvents service implementation to support resume #3283

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ require (
github.com/hyperledger/fabric-chaincode-go v0.0.0-20201119163726-f8ef75b17719
github.com/hyperledger/fabric-config v0.1.0
github.com/hyperledger/fabric-lib-go v1.0.0
github.com/hyperledger/fabric-protos-go v0.0.0-20220125190318-19041b215616
github.com/hyperledger/fabric-protos-go v0.0.0-20220315113721-7dc293e117f7
github.com/kr/pretty v0.3.0
github.com/miekg/pkcs11 v1.1.1
github.com/mitchellh/mapstructure v1.3.2
Expand Down
4 changes: 2 additions & 2 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,8 +239,8 @@ github.com/hyperledger/fabric-lib-go v1.0.0/go.mod h1:H362nMlunurmHwkYqR5uHL2UDW
github.com/hyperledger/fabric-protos-go v0.0.0-20190919234611-2a87503ac7c9/go.mod h1:xVYTjK4DtZRBxZ2D9aE4y6AbLaPwue2o/criQyQbVD0=
github.com/hyperledger/fabric-protos-go v0.0.0-20200424173316-dd554ba3746e/go.mod h1:xVYTjK4DtZRBxZ2D9aE4y6AbLaPwue2o/criQyQbVD0=
github.com/hyperledger/fabric-protos-go v0.0.0-20210911123859-041d13f0980c/go.mod h1:xVYTjK4DtZRBxZ2D9aE4y6AbLaPwue2o/criQyQbVD0=
github.com/hyperledger/fabric-protos-go v0.0.0-20220125190318-19041b215616 h1:CZrcDuLxBorn/xvbQl/r9kC0pniDEm0GuiBn9GgfY3c=
github.com/hyperledger/fabric-protos-go v0.0.0-20220125190318-19041b215616/go.mod h1:xVYTjK4DtZRBxZ2D9aE4y6AbLaPwue2o/criQyQbVD0=
github.com/hyperledger/fabric-protos-go v0.0.0-20220315113721-7dc293e117f7 h1:YV+siZuYQZwENjRH00t7ZS0CTlywt8Qog/SzL/jf6kE=
github.com/hyperledger/fabric-protos-go v0.0.0-20220315113721-7dc293e117f7/go.mod h1:xVYTjK4DtZRBxZ2D9aE4y6AbLaPwue2o/criQyQbVD0=
github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=
Expand Down
51 changes: 46 additions & 5 deletions integration/gateway/gateway_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -211,6 +211,7 @@ var _ = Describe("GatewayService", func() {
chaincodeEvents := func(
ctx context.Context,
startPosition *orderer.SeekPosition,
afterTxID string,
identity func() ([]byte, error),
sign func(msg []byte) ([]byte, error),
) (gateway.Gateway_ChaincodeEventsClient, error) {
Expand All @@ -225,6 +226,9 @@ var _ = Describe("GatewayService", func() {
if startPosition != nil {
request.StartPosition = startPosition
}
if len(afterTxID) > 0 {
request.AfterTransactionId = afterTxID
}

requestBytes, err := proto.Marshal(request)
Expect(err).NotTo(HaveOccurred())
Expand Down Expand Up @@ -332,7 +336,7 @@ var _ = Describe("GatewayService", func() {
},
}

eventsClient, err := chaincodeEvents(eventCtx, startPosition, signingIdentity.Serialize, signingIdentity.Sign)
eventsClient, err := chaincodeEvents(eventCtx, startPosition, "", signingIdentity.Serialize, signingIdentity.Sign)
Expect(err).NotTo(HaveOccurred())

_, transactionID := submitTransaction("event", []byte("EVENT_NAME"), []byte("EVENT_PAYLOAD"))
Expand Down Expand Up @@ -366,7 +370,7 @@ var _ = Describe("GatewayService", func() {
},
}

eventsClient, err := chaincodeEvents(eventCtx, startPosition, signingIdentity.Serialize, signingIdentity.Sign)
eventsClient, err := chaincodeEvents(eventCtx, startPosition, "", signingIdentity.Serialize, signingIdentity.Sign)
Expect(err).NotTo(HaveOccurred())

event, err := eventsClient.Recv()
Expand All @@ -383,13 +387,50 @@ var _ = Describe("GatewayService", func() {
Expect(proto.Equal(event.Events[0], expectedEvent)).To(BeTrue(), "Expected\n\t%#v\nto proto.Equal\n\t%#v", event.Events[0], expectedEvent)
})

It("should respond with replayed chaincode events after specified transaction ID", func() {
_, afterTransactionID := submitTransaction("event", []byte("WRONG_EVENT_NAME"), []byte("WRONG_EVENT_PAYLOAD"))
_, nextTransactionID := submitTransaction("event", []byte("CORRECT_EVENT_NAME"), []byte("CORRECT_EVENT_PAYLOAD"))

statusResult, err := commitStatus(afterTransactionID, signingIdentity.Serialize, signingIdentity.Sign)
Expect(err).NotTo(HaveOccurred())

_, err = commitStatus(nextTransactionID, signingIdentity.Serialize, signingIdentity.Sign)
Expect(err).NotTo(HaveOccurred())

eventCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()

startPosition := &orderer.SeekPosition{
Type: &orderer.SeekPosition_Specified{
Specified: &orderer.SeekSpecified{
Number: statusResult.BlockNumber,
},
},
}

eventsClient, err := chaincodeEvents(eventCtx, startPosition, afterTransactionID, signingIdentity.Serialize, signingIdentity.Sign)
Expect(err).NotTo(HaveOccurred())

event, err := eventsClient.Recv()
Expect(err).NotTo(HaveOccurred())

Expect(event.Events).To(HaveLen(1), "number of events")
expectedEvent := &peer.ChaincodeEvent{
ChaincodeId: "gatewaycc",
TxId: nextTransactionID,
EventName: "CORRECT_EVENT_NAME",
Payload: []byte("CORRECT_EVENT_PAYLOAD"),
}
Expect(proto.Equal(event.Events[0], expectedEvent)).To(BeTrue(), "Expected\n\t%#v\nto proto.Equal\n\t%#v", event.Events[0], expectedEvent)
})

It("should default to next commit if start position not specified", func() {
eventCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()

var startPosition *orderer.SeekPosition

eventsClient, err := chaincodeEvents(eventCtx, startPosition, signingIdentity.Serialize, signingIdentity.Sign)
eventsClient, err := chaincodeEvents(eventCtx, startPosition, "", signingIdentity.Serialize, signingIdentity.Sign)
Expect(err).NotTo(HaveOccurred())

_, transactionID := submitTransaction("event", []byte("EVENT_NAME"), []byte("EVENT_PAYLOAD"))
Expand Down Expand Up @@ -419,7 +460,7 @@ var _ = Describe("GatewayService", func() {
},
}

eventsClient, err := chaincodeEvents(eventCtx, startPosition, badIdentity.Serialize, signingIdentity.Sign)
eventsClient, err := chaincodeEvents(eventCtx, startPosition, "", badIdentity.Serialize, signingIdentity.Sign)
Expect(err).NotTo(HaveOccurred())

event, err := eventsClient.Recv()
Expand All @@ -443,7 +484,7 @@ var _ = Describe("GatewayService", func() {
},
}

eventsClient, err := chaincodeEvents(eventCtx, startPosition, signingIdentity.Serialize, badSign)
eventsClient, err := chaincodeEvents(eventCtx, startPosition, "", signingIdentity.Serialize, badSign)
Expect(err).NotTo(HaveOccurred())

event, err := eventsClient.Recv()
Expand Down
29 changes: 27 additions & 2 deletions internal/pkg/gateway/api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -534,6 +534,8 @@ func (gs *Server) ChaincodeEvents(signedRequest *gp.SignedChaincodeEventsRequest
return err
}

isMatch := chaincodeEventMatcher(request)

ledgerIter, err := ledger.GetBlocksIterator(startBlock)
if err != nil {
return status.Error(codes.Aborted, err.Error())
Expand All @@ -549,9 +551,8 @@ func (gs *Server) ChaincodeEvents(signedRequest *gp.SignedChaincodeEventsRequest
}

var matchingEvents []*peer.ChaincodeEvent

for _, event := range response.Events {
if event.GetChaincodeId() == request.GetChaincodeId() {
if isMatch(event) {
matchingEvents = append(matchingEvents, event)
}
}
Expand All @@ -572,6 +573,30 @@ func (gs *Server) ChaincodeEvents(signedRequest *gp.SignedChaincodeEventsRequest
}
}

func chaincodeEventMatcher(request *gp.ChaincodeEventsRequest) func(event *peer.ChaincodeEvent) bool {
chaincodeID := request.GetChaincodeId()
previousTransactionID := request.GetAfterTransactionId()

if len(previousTransactionID) == 0 {
return func(event *peer.ChaincodeEvent) bool {
return event.GetChaincodeId() == chaincodeID
}
}

passedPreviousTransaction := false

return func(event *peer.ChaincodeEvent) bool {
if !passedPreviousTransaction {
if event.TxId == previousTransactionID {
passedPreviousTransaction = true
}
return false
}

return event.GetChaincodeId() == chaincodeID
}
}

func startBlockFromLedgerPosition(ledger ledger.Ledger, position *ab.SeekPosition) (uint64, error) {
switch seek := position.GetType().(type) {
case nil:
Expand Down
Loading