Fix socket with file

SeaQL · Jul 4, 2023 · 21a11aa · 21a11aa
1 parent ac5fba9
commit 21a11aa
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 2 deletions.
diff --git a/sea-streamer-socket/src/streamer.rs b/sea-streamer-socket/src/streamer.rs
@@ -175,6 +175,12 @@ impl Streamer for SeaStreamer {
                         .await
                         .map_err(map_err)?,
                 ),
+                #[cfg(feature = "backend-file")]
+                "file" => SeaStreamerInner::File(
+                    FileStreamer::connect(uri, options.into_file_connect_options())
+                        .await
+                        .map_err(map_err)?,
+                ),
                 _ => {
                     return Err(StreamErr::Connect(format!("unknown protocol `{protocol}`")));
                 }

diff --git a/sea-streamer-types/src/streamer.rs b/sea-streamer-types/src/streamer.rs
@@ -112,8 +112,8 @@ impl StreamerUri {
     pub fn protocol(&self) -> Option<&str> {
         match self.nodes.first() {
             Some(node) => {
-                if !node.scheme().is_empty() && node.host().is_some() {
-                    Some(node.scheme())
+                if let Some((front, _)) = node.as_str().split_once("://") {
+                    Some(front)
                 } else {
                     None
                 }
@@ -338,6 +338,10 @@ mod test {
         let uri: StreamerUri = "file://./path/to/hi".parse().unwrap();
         assert_eq!(uri.protocol(), Some("file"));
         assert_eq!(uri.nodes(), &["file://./path/to/hi".parse().unwrap()]);
+
+        let uri: StreamerUri = "file:///path/to/hi".parse().unwrap();
+        assert_eq!(uri.protocol(), Some("file"));
+        assert_eq!(uri.nodes(), &["file:///path/to/hi".parse().unwrap()]);
     }
 
     #[test]