Make source tcp listener address configurable

CHANGELOG.md

@@ -14,6 +14,9 @@ All other sections are for end-users.
 
 - (Breaking Change) Remove `TimedStream` from foreign-service ([#250](https://github.com/SpringQL/SpringQL/pull/250)
 
+### Fixed
+- Make TCP server source address configurable, set it to "0.0.0.0" per default ([#260](https://github.com/SpringQL/SpringQL/pull/260))
+
 ## [v0.18.1] - 2022-10-07
 
 ### For developers

springql-core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "springql-core"
-version = "0.18.1"
+version = "0.18.2"
 
 authors = ["Sho Nakatani <lay.sakura@gmail.com>"]
 license = "MIT OR Apache-2.0"

springql-core/src/api/spring_config.rs

@@ -57,6 +57,7 @@ timeout_msec = 3_000
 [source_reader]
 net_connect_timeout_msec = 1_000
 net_read_timeout_msec = 100
+net_address = "0.0.0.0"
 
 can_read_timeout_msec = 100
 
@@ -175,10 +176,11 @@ pub struct SpringWebConsoleConfig {
 
 /// Config related to source reader
 #[allow(missing_docs)]
-#[derive(Copy, Clone, Eq, PartialEq, Debug, Deserialize)]
+#[derive(Clone, Eq, PartialEq, Debug, Deserialize)]
 pub struct SpringSourceReaderConfig {
     pub net_connect_timeout_msec: u32,
     pub net_read_timeout_msec: u32,
+    pub net_address: String,
 
     pub can_read_timeout_msec: u32,
 }

springql-core/src/stream_engine/autonomous_executor/repositories.rs

@@ -22,7 +22,7 @@ impl Repositories {
         Self {
             row_queue_repository: RowQueueRepository::default(),
             window_queue_repository: WindowQueueRepository::default(),
-            source_reader_repository: SourceReaderRepository::new(config.source_reader),
+            source_reader_repository: SourceReaderRepository::new(config.source_reader.clone()),
             sink_writer_repository: SinkWriterRepository::new(config.sink_writer),
         }
     }

springql-core/src/stream_engine/autonomous_executor/task/source_task/source_reader/net_server.rs

@@ -40,7 +40,7 @@ impl SourceReader for NetServerSourceReader {
             "unsupported protocol"
         );
 
-        let listener = TcpListener::bind(("127.0.0.1", options.port)).unwrap();
+        let listener = TcpListener::bind((config.net_address.clone(), options.port)).unwrap();
         let my_addr = listener.local_addr().unwrap();
 
         let (tx, rx) = mpsc::channel();
@@ -96,7 +96,10 @@ impl NetServerSourceReader {
     fn stream_handler(stream: TcpStream, tx: mpsc::Sender<serde_json::Value>) {
         log::info!(
             "[NetServerSourceReader] Connection from {}",
-            stream.peer_addr().unwrap()
+            stream
+                .peer_addr()
+                .map(|peer_addr| peer_addr.to_string())
+                .unwrap_or("unknown".to_string())
         );
 
         let mut tcp_reader = BufReader::new(stream);
@@ -128,7 +131,6 @@ impl NetServerSourceReader {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
     use crate::{
         api::SpringSinkWriterConfig,
         pipeline::OptionsBuilder,
@@ -137,6 +139,9 @@ mod tests {
             SchemalessRow,
         },
     };
+
+    use super::*;
+
     fn ephemeral_port() -> u16 {
         let addr = TcpListener::bind("127.0.0.1:0").unwrap();
         addr.local_addr().unwrap().port()