use crate::server::{BoundedWriter, LOG_TARGET};
use std::task::Poll;
use futures_util::{Future, FutureExt};
use http::Extensions;
use jsonrpsee_types::error::{
reject_too_big_batch_response, ErrorCode, ErrorObject, OVERSIZED_RESPONSE_CODE, OVERSIZED_RESPONSE_MSG,
};
use jsonrpsee_types::{ErrorObjectOwned, Id, Response, ResponsePayload as InnerResponsePayload};
use serde::Serialize;
use serde_json::value::to_raw_value;
#[derive(Debug, Clone)]
enum ResponseKind {
MethodCall,
Subscription,
Batch,
}
#[derive(Debug)]
pub struct MethodResponse {
result: String,
success_or_error: MethodResponseResult,
kind: ResponseKind,
on_close: Option<MethodResponseNotifyTx>,
extensions: Extensions,
}
impl MethodResponse {
pub fn is_success(&self) -> bool {
self.success_or_error.is_success()
}
pub fn is_error(&self) -> bool {
self.success_or_error.is_error()
}
pub fn is_subscription(&self) -> bool {
matches!(self.kind, ResponseKind::Subscription)
}
pub fn is_method_call(&self) -> bool {
matches!(self.kind, ResponseKind::MethodCall)
}
pub fn is_batch(&self) -> bool {
matches!(self.kind, ResponseKind::Batch)
}
pub fn into_result(self) -> String {
self.result
}
pub fn to_result(&self) -> String {
self.result.clone()
}
pub fn into_parts(self) -> (String, Option<MethodResponseNotifyTx>) {
(self.result, self.on_close)
}
pub fn as_error_code(&self) -> Option<i32> {
self.success_or_error.as_error_code()
}
pub fn as_result(&self) -> &str {
&self.result
}
pub fn from_batch(batch: BatchResponse) -> Self {
Self {
result: batch.0,
success_or_error: MethodResponseResult::Success,
kind: ResponseKind::Batch,
on_close: None,
extensions: Extensions::new(),
}
}
pub fn subscription_response<T>(id: Id, result: ResponsePayload<T>, max_response_size: usize) -> Self
where
T: Serialize + Clone,
{
let mut rp = Self::response(id, result, max_response_size);
rp.kind = ResponseKind::Subscription;
rp
}
pub fn response<T>(id: Id, rp: ResponsePayload<T>, max_response_size: usize) -> Self
where
T: Serialize + Clone,
{
let mut writer = BoundedWriter::new(max_response_size);
let success_or_error = if let InnerResponsePayload::Error(ref e) = rp.inner {
MethodResponseResult::Failed(e.code())
} else {
MethodResponseResult::Success
};
let kind = ResponseKind::MethodCall;
match serde_json::to_writer(&mut writer, &Response::new(rp.inner, id.clone())) {
Ok(_) => {
let result = unsafe { String::from_utf8_unchecked(writer.into_bytes()) };
Self { result, success_or_error, kind, on_close: rp.on_exit, extensions: Extensions::new() }
}
Err(err) => {
tracing::error!(target: LOG_TARGET, "Error serializing response: {:?}", err);
if err.is_io() {
let data = to_raw_value(&format!("Exceeded max limit of {max_response_size}")).ok();
let err_code = OVERSIZED_RESPONSE_CODE;
let err = InnerResponsePayload::<()>::error_borrowed(ErrorObject::borrowed(
err_code,
OVERSIZED_RESPONSE_MSG,
data.as_deref(),
));
let result =
serde_json::to_string(&Response::new(err, id)).expect("JSON serialization infallible; qed");
Self {
result,
success_or_error: MethodResponseResult::Failed(err_code),
kind,
on_close: rp.on_exit,
extensions: Extensions::new(),
}
} else {
let err = ErrorCode::InternalError;
let payload = jsonrpsee_types::ResponsePayload::<()>::error(err);
let result =
serde_json::to_string(&Response::new(payload, id)).expect("JSON serialization infallible; qed");
Self {
result,
success_or_error: MethodResponseResult::Failed(err.code()),
kind,
on_close: rp.on_exit,
extensions: Extensions::new(),
}
}
}
}
}
pub fn subscription_error<'a>(id: Id, err: impl Into<ErrorObject<'a>>) -> Self {
let mut rp = Self::error(id, err);
rp.kind = ResponseKind::Subscription;
rp
}
pub fn error<'a>(id: Id, err: impl Into<ErrorObject<'a>>) -> Self {
let err: ErrorObject = err.into();
let err_code = err.code();
let err = InnerResponsePayload::<()>::error_borrowed(err);
let result = serde_json::to_string(&Response::new(err, id)).expect("JSON serialization infallible; qed");
Self {
result,
success_or_error: MethodResponseResult::Failed(err_code),
kind: ResponseKind::MethodCall,
on_close: None,
extensions: Extensions::new(),
}
}
pub fn extensions(&self) -> &Extensions {
&self.extensions
}
pub fn extensions_mut(&mut self) -> &mut Extensions {
&mut self.extensions
}
pub fn with_extensions(self, extensions: Extensions) -> Self {
Self { extensions, ..self }
}
}
#[derive(Debug, Copy, Clone)]
enum MethodResponseResult {
Success,
Failed(i32),
}
impl MethodResponseResult {
fn is_success(&self) -> bool {
matches!(self, MethodResponseResult::Success)
}
fn is_error(&self) -> bool {
matches!(self, MethodResponseResult::Failed(_))
}
fn as_error_code(&self) -> Option<i32> {
match self {
Self::Failed(e) => Some(*e),
_ => None,
}
}
}
#[derive(Debug, Clone, Default)]
pub struct BatchResponseBuilder {
result: String,
max_response_size: usize,
}
impl BatchResponseBuilder {
pub fn new_with_limit(limit: usize) -> Self {
let mut initial = String::with_capacity(2048);
initial.push('[');
Self { result: initial, max_response_size: limit }
}
pub fn append(&mut self, response: &MethodResponse) -> Result<(), MethodResponse> {
let len = response.result.len() + self.result.len() + 1;
if len > self.max_response_size {
Err(MethodResponse::error(Id::Null, reject_too_big_batch_response(self.max_response_size)))
} else {
self.result.push_str(&response.result);
self.result.push(',');
Ok(())
}
}
pub fn is_empty(&self) -> bool {
self.result.len() <= 1
}
pub fn finish(mut self) -> BatchResponse {
if self.result.len() == 1 {
BatchResponse(batch_response_error(Id::Null, ErrorObject::from(ErrorCode::InvalidRequest)))
} else {
self.result.pop();
self.result.push(']');
BatchResponse(self.result)
}
}
}
#[derive(Debug, Clone)]
pub struct BatchResponse(String);
pub fn batch_response_error(id: Id, err: impl Into<ErrorObject<'static>>) -> String {
let err = InnerResponsePayload::<()>::error_borrowed(err);
serde_json::to_string(&Response::new(err, id)).expect("JSON serialization infallible; qed")
}
#[derive(Debug)]
pub struct ResponsePayload<'a, T>
where
T: Clone,
{
inner: InnerResponsePayload<'a, T>,
on_exit: Option<MethodResponseNotifyTx>,
}
impl<'a, T: Clone> From<InnerResponsePayload<'a, T>> for ResponsePayload<'a, T> {
fn from(inner: InnerResponsePayload<'a, T>) -> Self {
Self { inner, on_exit: None }
}
}
impl<'a, T> ResponsePayload<'a, T>
where
T: Clone,
{
pub fn success(t: T) -> Self {
InnerResponsePayload::success(t).into()
}
pub fn success_borrowed(t: &'a T) -> Self {
InnerResponsePayload::success_borrowed(t).into()
}
pub fn error(e: impl Into<ErrorObjectOwned>) -> Self {
InnerResponsePayload::error(e.into()).into()
}
pub fn error_borrowed(e: impl Into<ErrorObject<'a>>) -> Self {
InnerResponsePayload::error_borrowed(e.into()).into()
}
pub fn notify_on_completion(mut self) -> (Self, MethodResponseFuture) {
let (tx, rx) = response_channel();
self.on_exit = Some(tx);
(self, rx)
}
pub fn into_owned(self) -> ResponsePayload<'static, T> {
ResponsePayload { inner: self.inner.into_owned(), on_exit: self.on_exit }
}
}
impl<'a, T> From<ErrorCode> for ResponsePayload<'a, T>
where
T: Clone,
{
fn from(code: ErrorCode) -> Self {
let err: ErrorObject = code.into();
Self::error(err)
}
}
fn response_channel() -> (MethodResponseNotifyTx, MethodResponseFuture) {
let (tx, rx) = tokio::sync::oneshot::channel();
(MethodResponseNotifyTx(tx), MethodResponseFuture(rx))
}
#[derive(Debug)]
pub struct MethodResponseNotifyTx(tokio::sync::oneshot::Sender<NotifyMsg>);
impl MethodResponseNotifyTx {
pub fn notify(self, is_success: bool) {
let msg = if is_success { NotifyMsg::Ok } else { NotifyMsg::Err };
_ = self.0.send(msg);
}
}
#[derive(Debug)]
pub struct MethodResponseFuture(tokio::sync::oneshot::Receiver<NotifyMsg>);
#[derive(Debug, Copy, Clone)]
pub enum NotifyMsg {
Ok,
Err,
}
#[derive(Debug, Copy, Clone)]
pub enum MethodResponseError {
Closed,
JsonRpcError,
}
impl Future for MethodResponseFuture {
type Output = Result<(), MethodResponseError>;
fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
match self.0.poll_unpin(cx) {
Poll::Ready(Ok(NotifyMsg::Ok)) => Poll::Ready(Ok(())),
Poll::Ready(Ok(NotifyMsg::Err)) => Poll::Ready(Err(MethodResponseError::JsonRpcError)),
Poll::Ready(Err(_)) => Poll::Ready(Err(MethodResponseError::Closed)),
Poll::Pending => Poll::Pending,
}
}
}
#[cfg(test)]
mod tests {
use super::{BatchResponseBuilder, MethodResponse, ResponsePayload};
use jsonrpsee_types::Id;
#[test]
fn batch_with_single_works() {
let method = MethodResponse::response(Id::Number(1), ResponsePayload::success_borrowed(&"a"), usize::MAX);
assert_eq!(method.result.len(), 37);
let mut builder = BatchResponseBuilder::new_with_limit(39);
builder.append(&method).unwrap();
let batch = builder.finish();
assert_eq!(batch.0, r#"[{"jsonrpc":"2.0","result":"a","id":1}]"#)
}
#[test]
fn batch_with_multiple_works() {
let m1 = MethodResponse::response(Id::Number(1), ResponsePayload::success_borrowed(&"a"), usize::MAX);
assert_eq!(m1.result.len(), 37);
let limit = 2 + (37 * 2) + 1;
let mut builder = BatchResponseBuilder::new_with_limit(limit);
builder.append(&m1).unwrap();
builder.append(&m1).unwrap();
let batch = builder.finish();
assert_eq!(batch.0, r#"[{"jsonrpc":"2.0","result":"a","id":1},{"jsonrpc":"2.0","result":"a","id":1}]"#)
}
#[test]
fn batch_empty_err() {
let batch = BatchResponseBuilder::new_with_limit(1024).finish();
let exp_err = r#"{"jsonrpc":"2.0","error":{"code":-32600,"message":"Invalid request"},"id":null}"#;
assert_eq!(batch.0, exp_err);
}
#[test]
fn batch_too_big() {
let method = MethodResponse::response(Id::Number(1), ResponsePayload::success_borrowed(&"a".repeat(28)), 128);
assert_eq!(method.result.len(), 64);
let batch = BatchResponseBuilder::new_with_limit(63).append(&method).unwrap_err();
let exp_err = r#"{"jsonrpc":"2.0","error":{"code":-32011,"message":"The batch response was too large","data":"Exceeded max limit of 63"},"id":null}"#;
assert_eq!(batch.result, exp_err);
}
}