Skip to content

[ISSUE #4113]🚀Add RemotingGeneral struct and message processing methods #4114

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

Merged
rocketmq-rust-bot merged 2 commits into from
Sep 23, 2025
Merged

[ISSUE #4113]🚀Add RemotingGeneral struct and message processing methods #4114

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
963737c
[ISSUE #4113]🚀Add RemotingGeneral struct and message processing methods
mxsm Sep 23, 2025
591589e
fix code style
mxsm Sep 23, 2025
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
67 changes: 67 additions & 0 deletions rocketmq-remoting/src/remoting.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,3 +69,70 @@ pub trait InvokeCallback {
fn operation_succeed(&self, response: RemotingCommand);
fn operation_fail(&self, throwable: Box<dyn std::error::Error>);
}

#[allow(unused_variables)]
Copy link

Copilot AI Sep 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The #[allow(unused_variables)] attribute is applied broadly to the entire module. This should be removed once the empty method implementations are completed, as it may hide legitimate unused variable warnings in future development.

Suggested change
#[allow(unused_variables)]

Copilot uses AI. Check for mistakes.
mod inner {
use crate::base::response_future::ResponseFuture;
use crate::net::channel::Channel;
use crate::protocol::remoting_command::RemotingCommand;
use crate::protocol::RemotingCommandType;
use crate::remoting_server::server::Shutdown;
use crate::runtime::connection_handler_context::ConnectionHandlerContext;
use crate::runtime::processor::RequestProcessor;
use crate::runtime::RPCHook;
use std::collections::HashMap;
use std::sync::Arc;

struct RemotingGeneral<RP> {
request_processor: RP,
shutdown: Shutdown,
rpc_hooks: Arc<Vec<Box<dyn RPCHook>>>,
response_table: HashMap<i32, ResponseFuture>,
}

impl<RP> RemotingGeneral<RP> where RP: RequestProcessor + Sync + 'static + Clone {

pub fn process_message_received(&mut self, channel: Channel,
ctx: ConnectionHandlerContext,
cmd: &mut RemotingCommand,) {
match cmd.get_type() {
RemotingCommandType::REQUEST => {
self.process_request_command(channel, ctx, cmd);
}
RemotingCommandType::RESPONSE => {
self.process_response_command(channel, ctx, cmd);
}
}

}

fn process_request_command(&mut self, channel: Channel,
ctx: ConnectionHandlerContext,
cmd: &mut RemotingCommand,) {

}

fn process_response_command(&mut self, channel: Channel,
ctx: ConnectionHandlerContext,
cmd: &mut RemotingCommand,) {

}
Copy link

Copilot AI Sep 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The process_request_command and process_response_command methods are empty implementations. Consider adding TODO comments or basic error handling to indicate their intended functionality, or implement basic logging to track method invocation.

Copilot uses AI. Check for mistakes.

fn do_after_rpc_hooks(
&self,
channel: &Channel,
request: &RemotingCommand,
response: Option<&mut RemotingCommand>,
) -> rocketmq_error::RocketMQResult<()> {
unimplemented!("do_after_rpc_hooks unimplemented")
}

pub fn do_before_rpc_hooks(
&self,
channel: &Channel,
request: Option<&mut RemotingCommand>,
) -> rocketmq_error::RocketMQResult<()> {
unimplemented!("do_before_rpc_hooks unimplemented")
Comment on lines +136 to +144
Copy link

Copilot AI Sep 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Both RPC hook methods use unimplemented!() which will panic when called. Consider using todo!() instead to indicate these are placeholder implementations, or implement basic no-op functionality that returns Ok(()) to prevent runtime panics.

Copilot uses AI. Check for mistakes.
}
Copy link
Contributor

@coderabbitai coderabbitai bot Sep 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Avoid runtime panics: implement RPC hook methods instead of unimplemented!

These will panic if called. Mirror the existing server-side logic to make them no-ops when no hook is present and to uniformly execute hooks when provided.

Apply this diff:

-         fn do_after_rpc_hooks(
+         fn do_after_rpc_hooks(
             &self,
             channel: &Channel,
             request: &RemotingCommand,
             response: Option<&mut RemotingCommand>,
         ) -> rocketmq_error::RocketMQResult<()> {
-            unimplemented!("do_after_rpc_hooks unimplemented")
+            if let Some(response) = response {
+                for hook in self.rpc_hooks.iter() {
+                    hook.do_after_response(channel.remote_address(), request, response)?;
+                }
+            }
+            Ok(())
         }
 
-        pub fn do_before_rpc_hooks(
+        pub fn do_before_rpc_hooks(
             &self,
             channel: &Channel,
             request: Option<&mut RemotingCommand>,
         ) -> rocketmq_error::RocketMQResult<()> {
-            unimplemented!("do_before_rpc_hooks unimplemented")
+            if let Some(request) = request {
+                for hook in self.rpc_hooks.iter() {
+                    hook.do_before_request(channel.remote_address(), request)?;
+                }
+            }
+            Ok(())
         }
📝 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. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
fn do_after_rpc_hooks(
&self,
channel: &Channel,
request: &RemotingCommand,
response: Option<&mut RemotingCommand>,
) -> rocketmq_error::RocketMQResult<()> {
unimplemented!("do_after_rpc_hooks unimplemented")
}
pub fn do_before_rpc_hooks(
&self,
channel: &Channel,
request: Option<&mut RemotingCommand>,
) -> rocketmq_error::RocketMQResult<()> {
unimplemented!("do_before_rpc_hooks unimplemented")
}
fn do_after_rpc_hooks(
&self,
channel: &Channel,
request: &RemotingCommand,
response: Option<&mut RemotingCommand>,
) -> rocketmq_error::RocketMQResult<()> {
if let Some(response) = response {
for hook in self.rpc_hooks.iter() {
hook.do_after_response(channel.remote_address(), request, response)?;
}
}
Ok(())
}
pub fn do_before_rpc_hooks(
&self,
channel: &Channel,
request: Option<&mut RemotingCommand>,
) -> rocketmq_error::RocketMQResult<()> {
if let Some(request) = request {
for hook in self.rpc_hooks.iter() {
hook.do_before_request(channel.remote_address(), request)?;
}
}
Ok(())
}
🤖 Prompt for AI Agents 
In rocketmq-remoting/src/remoting.rs around lines 121 to 136, replace the
unimplemented! stubs for do_after_rpc_hooks and do_before_rpc_hooks with real
implementations that mirror the server-side behavior: check whether a configured
RPC hook exists and, if not, return Ok(()); if a hook is present, invoke the
appropriate hook method (passing the channel and request/response objects as the
server-side implementation does), handle any errors by converting/propagating
them into rocketmq_error::RocketMQResult, and ensure both methods return Ok(())
on success so they no longer panic at runtime.

}
}
Loading