Batching with array

.github/workflows/rust.yml

@@ -1,6 +1,10 @@
 name: CI
 
-on: [push, pull_request]
+on:
+  push:
+    branches:
+      - new-index
+  pull_request: {}
 
 jobs:
   test:
@@ -41,9 +45,13 @@ jobs:
     - name: Tests (Bitcoin mode, REST+Electrum)
       run: RUST_LOG=debug cargo test
 
+    - name: Test test_electrum_raw
+      run: RUST_LOG=debug cargo test -- --include-ignored test_electrum_raw
+
     - name: Tests (Liquid mode, REST)
       run: RUST_LOG=debug cargo test --features liquid
 
+
   nix:
     runs-on: ubuntu-latest
     steps:

src/electrum/server.rs

@@ -30,6 +30,7 @@ use crate::util::{create_socket, spawn_thread, BlockId, BoolThen, Channel, FullH
 const ELECTRS_VERSION: &str = env!("CARGO_PKG_VERSION");
 const PROTOCOL_VERSION: ProtocolVersion = ProtocolVersion::new(1, 4);
 const MAX_HEADERS: usize = 2016;
+const MAX_ARRAY_BATCH: usize = 20;
 
 #[cfg(feature = "electrum-discovery")]
 use crate::electrum::{DiscoveryManager, ServerFeatures};
@@ -541,53 +542,27 @@ impl Connection {
         let empty_params = json!([]);
         loop {
             let msg = receiver.recv().chain_err(|| "channel closed")?;
-            let start_time = Instant::now();
             trace!("RPC {:?}", msg);
             match msg {
                 Message::Request(line) => {
                     let cmd: Value = from_str(&line).chain_err(|| "invalid JSON format")?;
-                    match (
-                        cmd.get("method"),
-                        cmd.get("params").unwrap_or_else(|| &empty_params),
-                        cmd.get("id"),
-                    ) {
-                        (
-                            Some(&Value::String(ref method)),
-                            &Value::Array(ref params),
-                            Some(ref id),
-                        ) => {
-                            conditionally_log_rpc_event!(
-                                self,
-                                json!({
-                                    "event": "rpc request",
-                                    "id": id,
-                                    "method": method,
-                                    "params": if let Some(RpcLogging::Full) = self.rpc_logging {
-                                        json!(params)
-                                    } else {
-                                        Value::Null
-                                    }
-                                })
+                    if let Value::Array(arr) = cmd {
+                        if arr.len() > MAX_ARRAY_BATCH {
+                            bail!(
+                                "Too many elements in batch requests {} max:{}",
+                                arr.len(),
+                                MAX_ARRAY_BATCH
                             );
-
-                            let reply = self.handle_command(method, params, id)?;
-
-                            conditionally_log_rpc_event!(
-                                self,
-                                json!({
-                                    "event": "rpc response",
-                                    "method": method,
-                                    "payload_size": reply.to_string().as_bytes().len(),
-                                    "duration_micros": start_time.elapsed().as_micros(),
-                                    "id": id,
-                                })
-                            );
-
-                            self.send_values(&[reply])?
                         }
-                        _ => {
-                            bail!("invalid command: {}", cmd)
+                        let mut result = Vec::with_capacity(arr.len());
+                        for el in arr {
+                            let reply = self.handle_value(el, &empty_params)?;
+                            result.push(reply)
                         }
+                        self.send_values(&[Value::Array(result)])?
+                    } else {
+                        let reply = self.handle_value(cmd, &empty_params)?;
+                        self.send_values(&[reply])?
                     }
                 }
                 Message::PeriodicUpdate => {
@@ -601,6 +576,51 @@ impl Connection {
         }
     }
 
+    fn handle_value(&mut self, cmd: Value, empty_params: &Value) -> Result<Value> {
+        let start_time = Instant::now();
+        Ok(
+            match (
+                cmd.get("method"),
+                cmd.get("params").unwrap_or_else(|| empty_params),
+                cmd.get("id"),
+            ) {
+                (Some(&Value::String(ref method)), &Value::Array(ref params), Some(ref id)) => {
+                    conditionally_log_rpc_event!(
+                        self,
+                        json!({
+                            "event": "rpc request",
+                            "id": id,
+                            "method": method,
+                            "params": if let Some(RpcLogging::Full) = self.rpc_logging {
+                                json!(params)
+                            } else {
+                                Value::Null
+                            }
+                        })
+                    );
+
+                    let reply = self.handle_command(method, params, id)?;
+
+                    conditionally_log_rpc_event!(
+                        self,
+                        json!({
+                            "event": "rpc response",
+                            "method": method,
+                            "payload_size": reply.to_string().as_bytes().len(),
+                            "duration_micros": start_time.elapsed().as_micros(),
+                            "id": id,
+                        })
+                    );
+
+                    reply
+                }
+                _ => {
+                    bail!("invalid command: {}", cmd)
+                }
+            },
+        )
+    }
+
     fn parse_requests(mut reader: BufReader<TcpStream>, tx: &SyncSender<Message>) -> Result<()> {
         loop {
             let mut line = Vec::<u8>::new();

tests/electrum.rs

@@ -1,4 +1,7 @@
 pub mod common;
+use std::io::{Read, Write};
+use std::net::TcpStream;
+
 use common::Result;
 
 use bitcoind::bitcoincore_rpc::RpcApi;
@@ -137,3 +140,45 @@ fn test_electrum() -> Result<()> {
 
     Ok(())
 }
+
+/// Test the Electrum RPC server using an headless Electrum wallet
+/// This only runs on Bitcoin (non-Liquid) mode.
+#[cfg_attr(not(feature = "liquid"), test)]
+#[cfg_attr(feature = "liquid", allow(dead_code))]
+#[ignore = "must be launched singularly, otherwise conflict with the other server"]
+fn test_electrum_raw() {
+    // Spawn an Electrs Electrum RPC server
+    let (_electrum_server, electrum_addr, mut _tester) = common::init_electrum_tester().unwrap();
+    std::thread::sleep(std::time::Duration::from_millis(1000));
+
+    let mut stream = TcpStream::connect(electrum_addr).unwrap();
+    let write = "{\"jsonrpc\": \"2.0\", \"method\": \"server.version\", \"id\": 0}";
+
+    let s = write_and_read(&mut stream, write);
+    let expected = "{\"id\":0,\"jsonrpc\":\"2.0\",\"result\":[\"electrs-esplora 0.4.1\",\"1.4\"]}";
+    assert_eq!(s, expected);
+
+    let write = "[{\"jsonrpc\": \"2.0\", \"method\": \"server.version\", \"id\": 0}]";
+    let s = write_and_read(&mut stream, write);
+    let expected =
+        "[{\"id\":0,\"jsonrpc\":\"2.0\",\"result\":[\"electrs-esplora 0.4.1\",\"1.4\"]}]";
+    assert_eq!(s, expected);
+}
+
+fn write_and_read(stream: &mut TcpStream, write: &str) -> String {
+    stream.write_all(write.as_bytes()).unwrap();
+    stream.write(b"\n").unwrap();
+    stream.flush().unwrap();
+    let mut result = vec![];
+    loop {
+        let mut buf = [0u8];
+        stream.read_exact(&mut buf).unwrap();
+
+        if buf[0] == b'\n' {
+            break;
+        } else {
+            result.push(buf[0]);
+        }
+    }
+    std::str::from_utf8(&result).unwrap().to_string()
+}