Add Rust server and client for get_simple example (#8)

* Add Rust server and client for `get_simple` example * Remove and ignore `Cargo.lock` * Fix get request in client * Wrap span around the full example * Re-use the `BufReader` * Handle chunked transfer encoding in client * Apply suggestions from code review Co-authored-by: Ian Cook <ianmcook@gmail.com> --------- Co-authored-by: Ian Cook <ianmcook@gmail.com>
apache · Mar 11, 2024 · bbbfaf6 · bbbfaf6
1 parent 5657432
commit bbbfaf6
Show file tree

Hide file tree

Showing 8 changed files with 412 additions and 0 deletions.
diff --git a/http/get_simple/rs/.gitignore b/http/get_simple/rs/.gitignore
@@ -0,0 +1,19 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+/target
+Cargo.lock
diff --git a/http/get_simple/rs/Cargo.toml b/http/get_simple/rs/Cargo.toml
@@ -0,0 +1,27 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[workspace]
+resolver = "2"
+members = ["client", "server"]
+
+[workspace.dependencies]
+arrow-array = "50.0.0"
+arrow-ipc = "50.0.0"
+arrow-schema = "50.0.0"
+tracing = "0.1.40"
+tracing-subscriber = "0.3.18"
diff --git a/http/get_simple/rs/client/Cargo.toml b/http/get_simple/rs/client/Cargo.toml
@@ -0,0 +1,26 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "client"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+arrow-ipc.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
diff --git a/http/get_simple/rs/client/README.md b/http/get_simple/rs/client/README.md
@@ -0,0 +1,34 @@
+<!---
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing,
+  software distributed under the License is distributed on an
+  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  KIND, either express or implied.  See the License for the
+  specific language governing permissions and limitations
+  under the License.
+-->
+
+# HTTP GET Arrow Data: Simple Rust Client Example
+
+This directory contains a minimal example of an HTTP client implemented in Rust. The client:
+
+1. Sends an HTTP GET request to a server.
+2. Receives an HTTP 200 response from the server, with the response body containing an Arrow IPC stream of record batches.
+3. Adds the record batches to a list as they are received.
+
+To run this example, first start one of the server examples in the parent directory, then:
+
+```sh
+cargo r --release
+```
+> [!NOTE]  
+> This client example implements low-level HTTP/1.1 details directly, instead of using an HTTP library. We intend to update the example to use [hyper](https://docs.rs/hyper/latest/hyper/) after [arrow-rs has an async Arrow IPC reader](https://github.com/apache/arrow-rs/issues/1207)).
diff --git a/http/get_simple/rs/client/src/main.rs b/http/get_simple/rs/client/src/main.rs
@@ -0,0 +1,106 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use arrow_ipc::reader::StreamReader;
+use std::{
+    io::{BufRead, BufReader, Read, Write},
+    net::{IpAddr, Ipv4Addr, SocketAddr, TcpStream},
+};
+use tracing::{error, info, info_span};
+use tracing_subscriber::fmt::format::FmtSpan;
+
+fn main() {
+    // Configure tracing subscriber.
+    tracing_subscriber::fmt()
+        .with_span_events(FmtSpan::CLOSE)
+        .init();
+
+    info_span!("get_simple").in_scope(|| {
+        // Connect to server.
+        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8000);
+        match TcpStream::connect(addr) {
+            Ok(mut stream) => {
+                info_span!("Reading Arrow IPC stream", %addr).in_scope(|| {
+                    info!("Connected");
+
+                    // Send request.
+                    stream
+                        .write_all(format!("GET / HTTP/1.1\r\nHost: {addr}\r\n\r\n").as_bytes())
+                        .unwrap();
+
+                    // Ignore response header.
+                    let mut reader = BufReader::new(&mut stream);
+                    let mut chunked = false;
+                    loop {
+                        let mut line = String::default();
+                        reader.read_line(&mut line).unwrap();
+                        if let Some(("transfer-encoding", "chunked")) = line
+                            .to_lowercase()
+                            .split_once(':')
+                            .map(|(key, value)| (key.trim(), value.trim()))
+                        {
+                            chunked = true;
+                        }
+                        if line == "\r\n" {
+                            break;
+                        }
+                    }
+
+                    // Read Arrow IPC stream
+                    let batches: Vec<_> = if chunked {
+                        let mut buffer = Vec::default();
+                        loop {
+                            // Chunk size
+                            let mut line = String::default();
+                            reader.read_line(&mut line).unwrap();
+                            let chunk_size = u64::from_str_radix(line.trim(), 16).unwrap();
+
+                            if chunk_size == 0 {
+                                // Terminating chunk
+                                break;
+                            } else {
+                                // Append chunk to buffer
+                                let mut chunk_reader = reader.take(chunk_size);
+                                chunk_reader.read_to_end(&mut buffer).unwrap();
+                                // Terminating CR-LF sequence
+                                reader = chunk_reader.into_inner();
+                                reader.read_line(&mut String::default()).unwrap();
+                            }
+                        }
+                        StreamReader::try_new_unbuffered(buffer.as_slice(), None)
+                            .unwrap()
+                            .flat_map(Result::ok)
+                            .collect()
+                    } else {
+                        StreamReader::try_new_unbuffered(reader, None)
+                            .unwrap()
+                            .flat_map(Result::ok)
+                            .collect()
+                    };
+
+                    info!(
+                        batches = batches.len(),
+                        rows = batches.iter().map(|rb| rb.num_rows()).sum::<usize>()
+                    );
+                });
+            }
+            Err(error) => {
+                error!(%error, "Connection failed")
+            }
+        }
+    })
+}
diff --git a/http/get_simple/rs/server/Cargo.toml b/http/get_simple/rs/server/Cargo.toml
@@ -0,0 +1,31 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "server"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+arrow-array.workspace = true
+arrow-ipc.workspace = true
+arrow-schema.workspace = true
+once_cell = "1.19.0"
+rand = "0.8.5"
+rayon = "1.9.0"
+tracing.workspace = true
+tracing-subscriber.workspace = true
diff --git a/http/get_simple/rs/server/README.md b/http/get_simple/rs/server/README.md
@@ -0,0 +1,34 @@
+<!---
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing,
+  software distributed under the License is distributed on an
+  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  KIND, either express or implied.  See the License for the
+  specific language governing permissions and limitations
+  under the License.
+-->
+
+# HTTP GET Arrow Data: Simple Rust Server Example
+
+This directory contains a minimal example of an HTTP server implemented in Rust. The server:
+
+1. Creates a list of record batches and populates it with synthesized data.
+2. Listens for HTTP requests from clients.
+3. Upon receiving a request, sends an HTTP 200 response with the body containing an Arrow IPC stream of record batches.
+
+To run this example:
+
+```sh
+cargo r --release
+```
+> [!NOTE]  
+> This server example implements low-level HTTP/1.1 details directly, instead of using an HTTP library. We intend to update the example to use [hyper](https://docs.rs/hyper/latest/hyper/) after [arrow-rs has an async Arrow IPC writer](https://github.com/apache/arrow-rs/issues/1207)).