+29

Cargo.lock

reviewed

··· 1146 1146 ] 1147 1147 1148 1148 [[package]] 1149 1149 + name = "futures" 1150 1150 + version = "0.3.31" 1151 1151 + source = "registry+https://github.com/rust-lang/crates.io-index" 1152 1152 + checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876" 1153 1153 + dependencies = [ 1154 1154 + "futures-channel", 1155 1155 + "futures-core", 1156 1156 + "futures-executor", 1157 1157 + "futures-io", 1158 1158 + "futures-sink", 1159 1159 + "futures-task", 1160 1160 + "futures-util", 1161 1161 + ] 1162 1162 + 1163 1163 + [[package]] 1149 1164 name = "futures-buffered" 1150 1165 version = "0.2.12" 1151 1166 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1165 1180 checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10" 1166 1181 dependencies = [ 1167 1182 "futures-core", 1183 1183 + "futures-sink", 1168 1184 ] 1169 1185 1170 1186 [[package]] ··· 1172 1188 version = "0.3.31" 1173 1189 source = "registry+https://github.com/rust-lang/crates.io-index" 1174 1190 checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" 1191 1191 + 1192 1192 + [[package]] 1193 1193 + name = "futures-executor" 1194 1194 + version = "0.3.31" 1195 1195 + source = "registry+https://github.com/rust-lang/crates.io-index" 1196 1196 + checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f" 1197 1197 + dependencies = [ 1198 1198 + "futures-core", 1199 1199 + "futures-task", 1200 1200 + "futures-util", 1201 1201 + ] 1175 1202 1176 1203 [[package]] 1177 1204 name = "futures-io" ··· 1221 1248 source = "registry+https://github.com/rust-lang/crates.io-index" 1222 1249 checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" 1223 1250 dependencies = [ 1251 1251 + "futures-channel", 1224 1252 "futures-core", 1225 1253 "futures-io", 1226 1254 "futures-macro", ··· 1953 1981 "chrono", 1954 1982 "cid", 1955 1983 "ed25519-dalek", 1984 1984 + "futures", 1956 1985 "getrandom 0.3.4", 1957 1986 "http", 1958 1987 "ipld-core",

+4

crates/jacquard-common/Cargo.toml

reviewed

··· 78 78 optional = true 79 79 features = ["arithmetic"] 80 80 81 81 + [dev-dependencies] 82 82 + tokio = { version = "1", features = ["macros", "rt"] } 83 83 + futures = "0.3" 84 84 + 81 85 [package.metadata.docs.rs] 82 86 features = [ "crypto-k256", "crypto-k256", "crypto-p256"]

+78

crates/jacquard-common/src/stream.rs

reviewed

··· 87 87 } 88 88 } 89 89 90 90 + use bytes::Bytes; 91 91 + 92 92 + /// Platform-agnostic byte stream abstraction 93 93 + pub struct ByteStream { 94 94 + inner: Box<dyn n0_future::Stream<Item = Result<Bytes, StreamError>>>, 95 95 + } 96 96 + 97 97 + impl ByteStream { 98 98 + /// Create a new byte stream from any compatible stream 99 99 + pub fn new<S>(stream: S) -> Self 100 100 + where 101 101 + S: n0_future::Stream<Item = Result<Bytes, StreamError>> + 'static, 102 102 + { 103 103 + Self { 104 104 + inner: Box::new(stream), 105 105 + } 106 106 + } 107 107 + 108 108 + /// Check if stream is known to be empty (always false for dynamic streams) 109 109 + pub fn is_empty(&self) -> bool { 110 110 + false 111 111 + } 112 112 + 113 113 + /// Convert into the inner boxed stream 114 114 + pub fn into_inner(self) -> Box<dyn n0_future::Stream<Item = Result<Bytes, StreamError>>> { 115 115 + self.inner 116 116 + } 117 117 + } 118 118 + 119 119 + impl fmt::Debug for ByteStream { 120 120 + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 121 121 + f.debug_struct("ByteStream").finish_non_exhaustive() 122 122 + } 123 123 + } 124 124 + 125 125 + /// Platform-agnostic byte sink abstraction 126 126 + pub struct ByteSink { 127 127 + inner: Box<dyn n0_future::Sink<Bytes, Error = StreamError>>, 128 128 + } 129 129 + 130 130 + impl ByteSink { 131 131 + /// Create a new byte sink from any compatible sink 132 132 + pub fn new<S>(sink: S) -> Self 133 133 + where 134 134 + S: n0_future::Sink<Bytes, Error = StreamError> + 'static, 135 135 + { 136 136 + Self { 137 137 + inner: Box::new(sink), 138 138 + } 139 139 + } 140 140 + 141 141 + /// Convert into the inner boxed sink 142 142 + pub fn into_inner(self) -> Box<dyn n0_future::Sink<Bytes, Error = StreamError>> { 143 143 + self.inner 144 144 + } 145 145 + } 146 146 + 147 147 + impl fmt::Debug for ByteSink { 148 148 + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 149 149 + f.debug_struct("ByteSink").finish_non_exhaustive() 150 150 + } 151 151 + } 152 152 + 90 153 #[cfg(test)] 91 154 mod tests { 92 155 use super::*; 156 156 + use bytes::Bytes; 93 157 94 158 #[test] 95 159 fn stream_error_carries_kind_and_source() { ··· 107 171 108 172 assert_eq!(err.kind(), &StreamErrorKind::Closed); 109 173 assert!(err.source().is_none()); 174 174 + } 175 175 + 176 176 + #[tokio::test] 177 177 + async fn byte_stream_can_be_created() { 178 178 + use futures::stream; 179 179 + 180 180 + let data = vec![ 181 181 + Ok(Bytes::from("hello")), 182 182 + Ok(Bytes::from(" world")), 183 183 + ]; 184 184 + let stream = stream::iter(data); 185 185 + 186 186 + let byte_stream = ByteStream::new(stream); 187 187 + assert!(!byte_stream.is_empty()); 110 188 } 111 189 }