[CT-644] instantiate grpc stream manager

[jayy04]

Changelist

[Describe or list the changes made in this PR]

Test Plan

[Describe how this PR was tested (if applicable)]

Author/Reviewer Checklist

• If this PR has changes that result in a different app state given the same prior state and transaction list, manually add the state-breaking label.
• If the PR has breaking postgres changes to the indexer add the indexer-postgres-breaking label.
• If this PR isn't state-breaking but has changes that modify behavior in PrepareProposal or ProcessProposal, manually add the label proposal-breaking.
• If this PR is one of many that implement a specific feature, manually label them all feature:[feature-name].
• If you wish to for mergify-bot to automatically create a PR to backport your change to a release branch, manually add the label backport/[branch-name].
• Manually add any of the following labels: refactor, chore, bug.

[linear]

CT-644 Instantiate full node stream manager

[coderabbitai]

Walkthrough

The overall change introduces a new entity and infrastructure to manage gRPC streaming within the application. This includes the creation of a GrpcStreamingManager to handle streaming subscriptions and messages, along with its integration into the application's main structure and the keeper component. The update spans across adding new structs, interfaces, and methods to facilitate gRPC streaming management, aiming to enhance real-time data communication and processing capabilities.

Changes

Files	Change Summary
protocol/app/app.go protocol/x/clob/keeper/keeper.go	Introduced GrpcStreamingManager for managing gRPC streaming, integrated into the App struct and the keeper with necessary functions and fields.
protocol/streaming/grpc/grpc_streaming_manager.go protocol/streaming/grpc/noop_streaming_manager.go protocol/streaming/grpc/types/manager.go	Introduced GrpcStreamingManager infrastructure: GrpcStreamingManagerImpl and NoopGrpcStreamingManager for managing gRPC streaming subscriptions, along with the GrpcStreamingManager interface defining methods for enabling streaming, subscribing to updates, and sending messages.

Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media?

Share

• X
• Mastodon
• Reddit
• LinkedIn

---

Tips

Chat

There are 3 ways to chat with CodeRabbit:

• Review comments: Directly reply to a review comment made by CodeRabbit. Example:
  • I pushed a fix in commit <commit_id>.
  • Generate unit-tests for this file.
• Files and specific lines of code (under the "Files changed" tab): Tag @coderabbitai in a new review comment at the desired location with your query. Examples:
  • @coderabbitai generate unit tests for this file.
  • @coderabbitai modularize this function.
• PR comments: Tag @coderabbitai in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
  • @coderabbitai generate interesting stats about this repository from git and render them as a table.
  • @coderabbitai show all the console.log statements in this repository.
  • @coderabbitai read src/utils.ts and generate unit tests.
  • @coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.

Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments.

CodeRabbit Commands (invoked as PR comments)

• @coderabbitai pause to pause the reviews on a PR.
• @coderabbitai resume to resume the paused reviews.
• @coderabbitai review to trigger a review. This is useful when automatic reviews are disabled for the repository.
• @coderabbitai resolve resolve all the CodeRabbit review comments.
• @coderabbitai help to get help.

Additionally, you can add @coderabbitai ignore anywhere in the PR description to prevent this PR from being reviewed.

CodeRabbit Configration File (.coderabbit.yaml)

• You can programmatically configure CodeRabbit by adding a .coderabbit.yaml file to the root of your repository.
• The JSON schema for the configuration file is available here.
• If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: # yaml-language-server: $schema=https://coderabbit.ai/integrations/coderabbit-overrides.v2.json

CodeRabbit Discord Community

Join our Discord Community to get help, request features, and share feedback.

[coderabbitai]

Review Status

Actionable comments generated: 3

Configuration used: CodeRabbit UI

Commits

Files that changed from the base of the PR and between 7c6b3d9 and 15d8847.

Files selected for processing (6)

• protocol/app/app.go (5 hunks)
• protocol/streaming/grpc/grpc_streaming_manager.go (1 hunks)
• protocol/streaming/grpc/noop_streaming_manager.go (1 hunks)
• protocol/streaming/grpc/types/manager.go (1 hunks)
• protocol/testutil/keeper/clob.go (2 hunks)
• protocol/x/clob/keeper/keeper.go (5 hunks)

Additional comments: 9

protocol/streaming/grpc/types/manager.go (1)

• 1-19: The interface GrpcStreamingManager is well-defined, providing a clear contract for managing gRPC streaming with methods for enabling/disabling streaming, subscribing to order book updates, and sending messages. The use of specific types from the clobtypes package indicates a clear focus on order book functionality, which is appropriate for this context.

protocol/streaming/grpc/noop_streaming_manager.go (1)

• 1-36: The NoopGrpcStreamingManager provides a no-operation implementation of the GrpcStreamingManager interface, serving as a default or placeholder where no streaming functionality is required. This implementation is straightforward and adheres to best practices for providing default behavior in modular systems.

protocol/x/clob/keeper/keeper.go (4)

• 17-17: The addition of the streamingtypes import is appropriate for integrating the GrpcStreamingManager functionality into the keeper.
• 46-46: The introduction of the streamingManager field of type streamingtypes.GrpcStreamingManager is a logical extension of the keeper's responsibilities to include gRPC streaming management.
• 88-88: Updating the NewKeeper function to include grpcStreamingManager as a parameter is a necessary change for initializing the keeper with a streaming manager instance.
• 144-146: The addition of the GetGrpcStreamingManager method provides a clean and straightforward way to access the streamingManager instance from the keeper.

protocol/testutil/keeper/clob.go (1)

• 218-218: The initialization of streaming.NewNoopGrpcStreamingManager() in the createClobKeeper function is a sensible addition for test utilities, allowing for the isolation of tests from the gRPC streaming functionality by using a no-operation streaming manager.

protocol/app/app.go (2)

• 202-204: The import statements for the GrpcStreamingManager and its types are correctly placed and follow Go conventions. However, ensure that the imported packages are used effectively within the file and that there are no unused imports after these changes.
• 305-307: The addition of GrpcStreamingManager to the App struct is consistent with the application's architecture, allowing for centralized management of gRPC streaming. This change is well-integrated and follows the pattern of other managers within the App struct.

[coderabbitai]

The initialization of GrpcStreamingManager using getGrpcStreamingManagerFromOptions is a flexible approach that allows for configuration-based instantiation. However, it's important to ensure that the options and flags used for determining the manager type are clearly documented and validated. Additionally, consider adding error handling or logging within getGrpcStreamingManagerFromOptions to handle potential issues during initialization.

- return streaming.NewGrpcStreamingManager()
+ manager, err := streaming.NewGrpcStreamingManager()
+ if err != nil {
+     logger.Error("Failed to initialize GrpcStreamingManager", "error", err)
+     return nil
+ }
+ return manager

---

Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.

Suggested change

app.GrpcStreamingManager = getGrpcStreamingManagerFromOptions(appFlags, appOpts, logger)

[coderabbitai]

The function getGrpcStreamingManagerFromOptions correctly returns a GrpcStreamingManager instance based on application flags. However, the current implementation defaults to a no-op manager without considering any specific flags for enabling streaming. This might be intentional for a phased rollout or testing, but it's crucial to implement and document the flag-based logic for enabling the streaming manager in future iterations. Additionally, the TODO comment about adding command-line flags should be addressed to provide clarity on the intended usage and configuration options.

Consider implementing the flag-based logic for initializing the GrpcStreamingManager and removing the TODO comment once resolved.

[coderabbitai]

The GrpcStreamingManagerImpl serves as the actual implementation of the GrpcStreamingManager interface, with Enabled indicating streaming is enabled. However, the Subscribe and SendMessages methods are currently stubbed with no implementation, suggesting that this is an incomplete implementation. It would be beneficial to add a TODO or FIXME comment to indicate that further development is needed to provide the actual streaming functionality.

Would you like me to add a TODO comment or open a GitHub issue to track the completion of this implementation?

[coderabbitai]

Review Status

Actionable comments generated: 0

Configuration used: CodeRabbit UI

Commits

Files that changed from the base of the PR and between 15d8847 and fe41e97.

Files selected for processing (3)

• protocol/streaming/grpc/grpc_streaming_manager.go (1 hunks)
• protocol/streaming/grpc/noop_streaming_manager.go (1 hunks)
• protocol/streaming/grpc/types/manager.go (1 hunks)

Files skipped from review as they are similar to previous changes (3)

• protocol/streaming/grpc/grpc_streaming_manager.go
• protocol/streaming/grpc/noop_streaming_manager.go
• protocol/streaming/grpc/types/manager.go

[coderabbitai]

Review Status

Actionable comments generated: 0

Configuration used: CodeRabbit UI

Commits

Files that changed from the base of the PR and between fe41e97 and 0d40546.

Files selected for processing (3)

• protocol/streaming/grpc/grpc_streaming_manager.go (1 hunks)
• protocol/streaming/grpc/noop_streaming_manager.go (1 hunks)
• protocol/streaming/grpc/types/manager.go (1 hunks)

Files skipped from review as they are similar to previous changes (3)

• protocol/streaming/grpc/grpc_streaming_manager.go
• protocol/streaming/grpc/noop_streaming_manager.go
• protocol/streaming/grpc/types/manager.go

[jonfung-dydx]

maybe add a comment about lifecycles of the req, srv, finished variables. i think there's a stream and a chan going on so not too sure when the finished chan returns a bool.