some strange problems

[vCassius]

i use this server.start(module).stopped().await; to keep service running with ws model. but sometimes service closed without any error info. tellnet ws port fail.

[niklasad1]

Hey @Cassuis,

Can you explain which jsonrpsee version and what's the behaviour when the "service closes" i.e, does the server just refuse the connection or what is going on?

[vCassius]

version: jsonrpsee = { version = "0.20" , features = ["server", "http-client", "ws-client", "macros", "client-ws-transport-native-tls"]}
behaviour : refuse the connection ,port not bind.

[niklasad1]

Hmm, so that once you start the server the telnet keep alive check works and after a while it stops working with the behavior as you stated 👆?

Any way to reproduce that locally and can you please share the code you are running to trigger this behavior?

[vCassius]

here is my source!~

pub async fn start() {
    let Some(conf) = INSTANCE_CONF.get() else {
        INSTANCE_CONF.get();
        exit(-1)
    };
    let cors = CorsLayer::new()
        .allow_methods([Method::POST])
        .allow_origin(Any)
        .allow_headers([hyper::header::CONTENT_TYPE]);
    let middleware = tower::ServiceBuilder::new().layer(cors);
    let addrs = format!("{}:{}", conf.config.rpc.ip, conf.config.rpc.port);
    let service_address = match addrs.parse::<SocketAddr>() {
        Ok(s) => s,
        Err(e) => {
            tracing::error!("fail to get server address err: {e}");
            exit(-1)
        }
    };
    let server = match Server::builder()
        .set_middleware(middleware)
        .build(service_address)
        .await
    {
        Ok(s) => s,
        Err(e) => {
            tracing::error!("fail to get build server err: {e}");
            exit(-1)
        }
    };
    let mut module = RpcModule::new(());
    match module.register_async_method("say_hello", move |params, _| async move {
        pkg::rpc::handler::test::hello::say(params).await
    }) {
        Ok(_) => {}
        Err(e) => {
            tracing::error!("fail to register method say_hello,err: {e}");
            exit(-1)
        }
    };
    match module.register_async_method("ConfigMap.List", |params, _| async move {
        pkg::rpc::handler::kubernetes::configmap::list(params).await
    }) {
        Ok(c) => {
            tracing::info!("configmap list register succeed,call back: {:?}", c)
        }
        Err(e) => {
            let err = format!("fail to register method configmap list,err: {e}");
            send(SimpleMessage {
                title: "rpc service exception".to_string(),
                message: err.clone(),
                color: TITLE_TEMPLATE_COLOR_RED.to_string(),
            })
            .await;
            tracing::error!(err);
            exit(-1)
        }
    };
    match module.register_async_method("Deployment.List", |params, _| async {
        pkg::rpc::handler::kubernetes::deployment::list(params).await
    }) {
        Ok(d) => {
            tracing::info!("deployment list register succeed,call back: {:?}", d)
        }
        Err(e) => {
            let err = format!("fail to register method deployment list,err: {e}");
            send(SimpleMessage {
                title: "rpc service exception".to_string(),
                message: err.clone(),
                color: TITLE_TEMPLATE_COLOR_RED.to_string(),
            })
            .await;
            tracing::error!(err);
            exit(-1)
        }
    };
    let addr = match server.local_addr() {
        Ok(a) => a,
        Err(e) => {
            let err = format!("fail to get socket address err: {e}");
            send(SimpleMessage {
                title: "rpc service exception".to_string(),
                message: err.clone(),
                color: TITLE_TEMPLATE_COLOR_RED.to_string(),
            })
            .await;
            tracing::error!(err);
            exit(-1)
        }
    };
    tracing::info!("rpc server is running on {}", addr);
    server.start(module).stopped().await;
}

and called list fn

pub async fn list(params: Params<'_>) -> RpcResult<DeploymentList> {
    let param: JsonValue = params.parse()?;
    let js = param.to_string();
    let obj = match pkg::utils::serialize::deserialize_json::<Vec<List>>(js.as_str()).await {
        Ok(o) => o,
        Err(e) => {
            tracing::error!("fail to deserialize,err: {e}");
            return Err(pkg::rpc::exception::error::message("fail to deserialize"));
        }
    };
    if !obj.is_empty() {
    } else {
        tracing::error!("request object is empty");
        return Err(pkg::rpc::exception::error::message(
            "request object is empty",
        ));
    }
    let Some(sender_pool) =  INSTANCE_CHANNEL_SENDER_POOL.get() else  {
        tracing::error!("fail too get sender pool");
        return Err(pkg::rpc::exception::error::message(
            "internal logic exception",
        ));
    };
    let Some(sender) = sender_pool
        .pool
        .get(obj[0].cluster.clone().as_str())else {
        let err = format!("fail to get sender channel with cluster label: {}",obj[0].cluster.clone());
        let msg_text = format!(
            "**rpc sender channel exception**\n### method: list object: configmap\n### message: {}",
            err,
        );
        send(SimpleMessage{
            title: "rpc service exception".to_string(),
            message: msg_text.to_string(),
            color: TITLE_TEMPLATE_COLOR_RED.to_string(),
        }).await;
        tracing::error!(err);
        return Err(pkg::rpc::exception::error::message(
            "internal logic exception",
        ));
    };
    let uuid = pkg::utils::uuid::get_without_async();
    let msg_data = message::MsgData {
        err: None,
        msg_uuid: uuid.clone(),
        action: Some(obj[0].clone().msg_data_action),
        mark_type: None,
        conn_des: None,
        data: Some(obj[0].clone()),
    };
    let data = match pkg::utils::serialize::_serialize_json(msg_data).await {
        Ok(d) => d,
        Err(e) => {
            tracing::error!("fail to _serialize to json,err: {e}");
            return Err(pkg::rpc::exception::error::message(
                "fail to serialize json",
            ));
        }
    };
    let m = Message {
        id: pkg::tcp::logic::protocol::PROTOCOL_DEPLOYMENT,
        data_len: 0,
        data,
    };
    if !sender.is_closed() {
        match sender.send(m).await {
            Ok(_) => {}
            Err(e) => {
                tracing::error!("fail to send message with channel,err: {e}");
                return Err(pkg::rpc::exception::error::message("fail to send message"));
            }
        };
    } else {
        tracing::error!("fail to send message with channel,cuz channel closed");
        return Err(pkg::rpc::exception::error::message(
            "fail to send message with channel,cuz channel closed",
        ));
    }
    let result = match message::parse_rev_msg_data::<DeploymentList>(uuid.clone()).await {
        Ok(m) => m,
        Err(e) => {
            tracing::error!("fail to parse deployment list return message,err: {e}");
            return Err(pkg::rpc::exception::error::message(format!("{e}")));
        }
    };
    Ok(result.data)
}

[niklasad1]

Ok,

I have noticed some weird issues with the graceful shutdown and my guess is what you run into, can you please try latest master if it fixes things for you?

I need to test before making another release.

[vCassius]

i found why,cuz i have a Hashmap(used crate Dashmap) to save message,and tcp modele use map to insert and delete timeout message,rpc request and select if message exit in map, when tcp model iter and remove key here's the problem,i have write a demo,it's not jsonrpsee's problem,but another question,i used golang jsonrcp 2.0 client ,how to use jsonrpsee server to response error message? seems code does not work like this

 return Err(pkg::rpc::exception::err::message("fail to deserialize"));


use jsonrpsee::types::ErrorObjectOwned;
use jsonrpsee::types::error::ErrorObjectOwned as JsonRpcErrorOwned;
use jsonrpsee::types::error::SERVER_ERROR_MSG;

pub fn message(e: impl ToString) -> ErrorObjectOwned {
    JsonRpcErrorOwned::owned(-32000, SERVER_ERROR_MSG, Some(e.to_string()))
}

[niklasad1]

Yeah, using the register_method can be a little tricky to work with because you need to specify the error type in the closure because you can also use your own error type implements Into<ErrorObject>.

Example code that you can base your code from

	module.register_method::<Result<usize, ErrorObjectOwned>, _>("multiply", |params, _| {
		let n: usize = params.parse()?;

		Ok(n * 2)
	})?;

	module.register_async_method::<Result<usize, ErrorObjectOwned>, _, _>(
		"multiply_async",
		|params, _| async move {
			let n: usize = params.parse()?;

			Ok(n * 2)
		},
	)?;

However, it would be easier for you if you use the proc macro API generation then you don't have to deal with it yourself:

use jsonrpsee::core::async_trait;
use jsonrpsee::proc_macros::rpc;

#[rpc(server)]
pub trait Api {
	#[method(name = "multiply")]
	fn multiply(&self, n: usize) -> Result<usize, ErrorObjectOwned>;

	#[method(name = "multiply_async")]
	async fn multiply_async(&self, n: usize) -> Result<usize, ErrorObjectOwned>;
}

struct RpcImpl;

#[async_trait]
impl ApiServer for RpcImpl {
	fn multiply(&self, n: usize) -> Result<usize, ErrorObjectOwned> {
		Ok(n * 2)
	}

	async fn multiply_async(&self, n: usize) -> Result<usize, ErrorObjectOwned> {
		Ok(n * 2)
	}
}

server.start(RpcImpl.into_rpc()).unwrap();

Then general guidelines is that you would have your own Error type that implements Into<ErrorObjectOwned>, we wrote some tips about in https://github.com/paritytech/jsonrpsee/blob/master/CHANGELOG.md#v0180---2023-04-21

[vCassius]

thanks alot for ur help! i will try to this!~