1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
use crate::io::{BufReader, BufWriter, Read, Write};
use crate::{
error::{IpcError, ProtocolErrorCode},
ipc::Serve,
packet::{Packet, RequestPacket, ResponsePacket},
};
use alloc::string::String;
use alloc::vec;
use serde::{Deserialize, Serialize};
use serde_json::{from_slice, to_vec};
/// The `Channel` struct facilitates communication between a client and a server.
/// It handles the transmission of requests from the client to the server and the reception
/// of responses from the server to the client. Communication is achieved through the use of pipes.
///
/// # Fields
///
/// * `reader` - Responsible for reading data from the channel.
/// * `writer` - Responsible for writing data to the channel.
pub struct Channel<R: Read, W: Write> {
reader: BufReader<R>,
writer: BufWriter<W>,
}
impl<R: Read, W: Write> Channel<R, W> {
pub fn new(reader: R, writer: W) -> Self {
Self {
reader: BufReader::new(reader),
writer: BufWriter::new(writer),
}
}
}
impl<R: Read, W: Write> Channel<R, W> {
/// Executes the server loop, processing incoming requests and sending responses.
///
/// This function runs an infinite loop, continuously receiving requests from the client,
/// processing them using the provided service implementation, and sending back the responses.
/// If an error occurs during the processing of a request or the sending of a response, the
/// error is logged (if logging is enabled) and the error code is sent back to the client.
///
/// # Arguments
///
/// * `serve` - A mutable reference to the service implementation that handles the requests and
/// generates the responses. The service must implement the `Serve` trait with the appropriate
/// request and response types.
///
/// # Type Parameters
///
/// * `Req` - The type of the request messages. It must implement `Serialize` and `Deserialize`.
/// * `Resp` - The type of the response messages. It must implement `Serialize` and `Deserialize`.
/// * `S` - The type of the service implementation. It must implement the `Serve` trait with
/// `Req` as the request type and `Resp` as the response type.
///
/// # Returns
///
/// A `Result` indicating the success or failure of the server execution. If the server runs
/// successfully, it never returns. If an error occurs, it returns an `IpcError`.
pub fn execute<Req, Resp, S>(mut self, serve: &mut S) -> Result<(), IpcError>
where
Req: Serialize + for<'de> Deserialize<'de>,
Resp: Serialize + for<'de> Deserialize<'de>,
S: Serve<Req = Req, Resp = Resp>,
{
loop {
let result = self
.receive_request()
.and_then(|req| serve.serve(req))
.and_then(|resp| self.send_response(resp));
match result {
Ok(_) => continue,
Err(e) => {
#[cfg(feature = "enable-logging")]
log::error!("Error in execute loop: {:?}", e);
// notify client
self.send_error_code(e.clone().into()).unwrap();
return Err(e);
}
}
}
}
///
/// Sends a request to the server and waits for a response.
///
/// This function serializes the request, sends it to the server, and then waits for the server's response.
/// It returns the deserialized response or an `IpcError` if an error occurs during the process.
///
/// # Arguments
///
/// * `_method_name` - A static string slice representing the name of the method being called.
/// * `req` - The request message to be sent to the server. It must implement `Serialize` and `Deserialize`.
///
/// # Type Parameters
///
/// * `Req` - The type of the request message. It must implement `Serialize` and `Deserialize`.
/// * `Resp` - The type of the response message. It must implement `Serialize` and `Deserialize`.
///
/// # Returns
///
/// A `Result` containing the deserialized response message if the call is successful, or an `IpcError` if
/// an error occurs during the process.
/// # Example
///
/// ```rust,ignore
/// use ckb_script_ipc_common::channel::Channel;
/// use serde::{Serialize, Deserialize};
///
/// #[derive(Serialize, Deserialize)]
/// struct MyRequest {
/// // request fields
/// }
///
/// #[derive(Serialize, Deserialize)]
/// struct MyResponse {
/// // response fields
/// }
///
/// let mut channel = Channel::new(reader, writer);
/// let request = MyRequest { /* fields */ };
/// let response: MyResponse = channel.call("my_method", request).expect("Failed to call method");
/// ```
pub fn call<Req, Resp>(
&mut self,
_method_name: &'static str,
req: Req,
) -> Result<Resp, IpcError>
where
Req: Serialize + for<'de> Deserialize<'de>,
Resp: Serialize + for<'de> Deserialize<'de>,
{
let result = self.send_request(req).and_then(|_| self.receive_response());
match result {
Ok(resp) => Ok(resp),
Err(e) => {
#[cfg(feature = "enable-logging")]
log::error!("Error in call({}): {:?}", _method_name, e);
Err(e)
}
}
}
pub(crate) fn send_request<Req: Serialize>(&mut self, req: Req) -> Result<(), IpcError> {
let serialized_req = to_vec(&req).map_err(|_| IpcError::SerializeError)?;
let packet = RequestPacket::new(serialized_req);
#[cfg(feature = "enable-logging")]
log::info!("send request: {:?}", packet);
let bytes = packet.serialize();
self.writer.write(&bytes)?;
self.writer.flush()?;
Ok(())
}
/// Sends a raw JSON string request to the server.
///
/// This function takes a JSON string and sends it directly as a request packet to the server,
/// without performing any serialization. This is useful when working with raw JSON data that
/// doesn't need to be converted from Rust types.
///
/// # Arguments
///
/// * `json` - A string slice containing the JSON request to be sent.
///
/// # Returns
///
/// A `Result` indicating whether the request was successfully sent, or an `IpcError` if
/// writing to the channel fails.
pub fn send_json_request(&mut self, json: &str) -> Result<(), IpcError> {
let packet = RequestPacket::new(json.as_bytes().to_vec());
#[cfg(feature = "enable-logging")]
log::info!("send request: {:?}", packet);
let bytes = packet.serialize();
self.writer.write(&bytes)?;
self.writer.flush()?;
Ok(())
}
pub(crate) fn send_response<Resp: Serialize>(&mut self, resp: Resp) -> Result<(), IpcError> {
let serialized_resp = to_vec(&resp).map_err(|_| IpcError::SerializeError)?;
let packet = ResponsePacket::new(0, serialized_resp);
#[cfg(feature = "enable-logging")]
log::info!("send response: {:?}", packet);
let bytes = packet.serialize();
self.writer.write(&bytes)?;
self.writer.flush()?;
Ok(())
}
pub(crate) fn send_error_code(
&mut self,
error_code: ProtocolErrorCode,
) -> Result<(), IpcError> {
let packet = ResponsePacket::new(error_code.clone() as u64, vec![]);
#[cfg(feature = "enable-logging")]
log::info!("send error code: {:?}", error_code as u64);
let bytes = packet.serialize();
self.writer.write(&bytes)?;
self.writer.flush()?;
Ok(())
}
pub(crate) fn receive_request<Req: for<'de> Deserialize<'de>>(
&mut self,
) -> Result<Req, IpcError> {
let packet = RequestPacket::read_from(&mut self.reader)?;
#[cfg(feature = "enable-logging")]
log::info!("receive request: {:?}", packet);
let req = from_slice(packet.payload()).map_err(|_| IpcError::DeserializeError)?;
Ok(req)
}
pub(crate) fn receive_response<Resp: for<'de> Deserialize<'de>>(
&mut self,
) -> Result<Resp, IpcError> {
let packet = ResponsePacket::read_from(&mut self.reader)?;
#[cfg(feature = "enable-logging")]
log::info!("Received response: {:?}", packet);
let error_code = ProtocolErrorCode::from(packet.error_code());
match error_code {
ProtocolErrorCode::Ok => {}
e => {
#[cfg(feature = "enable-logging")]
log::error!("Received error code: {:?}", e);
return Err(IpcError::ProtocolError(e));
}
}
from_slice(packet.payload()).map_err(|_| IpcError::DeserializeError)
}
/// Receives a JSON string response from the server.
///
/// This function reads a response packet from the server and returns its payload as a String,
/// without attempting to deserialize it into a specific type. This is useful when working
/// with raw JSON responses that don't need to be converted into specific Rust types.
///
/// # Returns
///
/// * `Ok(String)` - A String containing the JSON response if successful
/// * `Err(IpcError)` - An error if:
/// - Reading from the channel fails
/// - The server returns an error code
/// - The response payload contains invalid UTF-8
///
pub fn receive_json_response(&mut self) -> Result<String, IpcError> {
let packet = ResponsePacket::read_from(&mut self.reader)?;
#[cfg(feature = "enable-logging")]
log::info!("Received response: {:?}", packet);
let error_code = ProtocolErrorCode::from(packet.error_code());
match error_code {
ProtocolErrorCode::Ok => {}
e => {
#[cfg(feature = "enable-logging")]
log::error!("Received error code: {:?}", e);
return Err(IpcError::ProtocolError(e));
}
}
Ok(String::from_utf8_lossy(packet.payload()).into_owned())
}
}