ingest/ledgerbackend: Create functional producer for BufferedStorageBackend #5462
Conversation
| return | ||
| } | ||
| } | ||
| close(resultCh) |
There was a problem hiding this comment.
Why isn't the channel closed on early return due to an error?
| } | ||
| resultCh := make(chan error, 1) | ||
|
|
||
| go func() { |
There was a problem hiding this comment.
Should the context cancellation be handled in this goroutine to handle caller cancelling the context?
|
|
||
| err = callback(ledgerCloseMeta) | ||
| if err != nil { | ||
| resultCh <- errors.Wrap(err, "received an error from callback invocation") |
There was a problem hiding this comment.
nit: should we consistently use fmt.Errorf ?
| select { | ||
| case chErr, ok := <-resultCh: | ||
| if ok { | ||
| assert.ErrorContains(t, chErr, "invalid end value for unbounded ranged, must be zero") |
There was a problem hiding this comment.
| assert.ErrorContains(t, chErr, "invalid end value for unbounded ranged, must be zero") | |
| assert.ErrorContains(t, chErr, "invalid end value for unbounded range, must be zero") |
| } | ||
|
|
||
| if !ledgerRange.bounded && ledgerRange.to > 0 { | ||
| resultCh <- errors.New("invalid end value for unbounded ranged, must be zero") |
There was a problem hiding this comment.
| resultCh <- errors.New("invalid end value for unbounded ranged, must be zero") | |
| resultCh <- errors.New("invalid end value for unbounded range, must be zero") |
| } | ||
| resultCh := make(chan error, 1) | ||
|
|
||
| go func() { |
There was a problem hiding this comment.
It seems to be a best practice to avoid asynchronous APIs in Go, see:
https://google.github.io/styleguide/go/decisions.html#synchronous-functions
https://stackoverflow.com/a/28510173
page 9 of https://drive.google.com/file/d/1nPdvhB0PutEJzdCq5ms6UI58dp50fcAN/view
| if !ledgerRange.bounded { | ||
| to = math.MaxUint32 | ||
| } |
There was a problem hiding this comment.
you can use the following for loop condition to avoid setting to to the max uint32:
for ledgerSeq := from; ledgerSeq <= to || !ledgerRange.bounded; ledgerSeq++ {
PR Checklist
PR Structure
otherwise).
services/friendbot, orallordocif the changes are broad or impact manypackages.
Thoroughness
.mdfiles, etc... affected by this change). Take a look in the
docsfolder for a given service,like this one.
Release planning
CHANGELOG.mdwithin the component folder structure. For example, if I changed horizon, then I updated (services/horizon/CHANGELOG.md. I add a new line item describing the change and reference to this PR. If I don't update a CHANGELOG, I acknowledge this PR's change may not be mentioned in future release notes.semver, or if it's mainly a patch change. The PR is targeted at the next
release branch if it's not a patch change.
What
Added new
PublishFromBufferedStorageBackendto provide a functional origin for streaming data development as part of CDP processors library.Why
Provide efficient tooling in the sdk to produce a stream of tx-meta from a remote Datastore that has pre-computed tx-meta files from Galexie. The stream is established as a callback function which caller passes, it will be asynchronously invoked and serves as origin in a data transformation pipeline, providing source of network tx-meta to which pipeline transform operators can derive from.
Closes: #5412
Known limitations