hydro_lang/live_collections/keyed_stream/networking.rs
1//! Networking APIs for [`KeyedStream`].
2
3use serde::Serialize;
4use serde::de::DeserializeOwned;
5use stageleft::{q, quote_type};
6
7use super::KeyedStream;
8use crate::compile::ir::{DebugInstantiate, HydroNode, NetworkRecv, NetworkSend};
9use crate::live_collections::boundedness::{Boundedness, Unbounded};
10use crate::live_collections::stream::{MinOrder, Ordering, Retries, Stream};
11use crate::location::cluster::{Consistency, NoConsistency};
12#[cfg(stageleft_runtime)]
13use crate::location::dynamic::DynLocation;
14use crate::location::{Cluster, MemberId, Process};
15use crate::networking::{NetworkFor, TCP};
16
17impl<'a, T, L, L2, B: Boundedness, O: Ordering, R: Retries>
18 KeyedStream<MemberId<L2>, T, Process<'a, L>, B, O, R>
19{
20 #[deprecated = "use KeyedStream::demux(..., TCP.fail_stop().bincode()) instead"]
21 /// Sends each group of this stream to a specific member of a cluster, with the [`MemberId`] key
22 /// identifying the recipient for each group and using [`bincode`] to serialize/deserialize messages.
23 ///
24 /// Each key must be a `MemberId<L2>` and each value must be a `T` where the key specifies
25 /// which cluster member should receive the data. Unlike [`Stream::broadcast_bincode`], this
26 /// API allows precise targeting of specific cluster members rather than broadcasting to
27 /// all members.
28 ///
29 /// # Example
30 /// ```rust
31 /// # #[cfg(feature = "deploy")] {
32 /// # use hydro_lang::prelude::*;
33 /// # use futures::StreamExt;
34 /// # tokio_test::block_on(hydro_lang::test_util::multi_location_test(|flow, p2| {
35 /// let p1 = flow.process::<()>();
36 /// let workers: Cluster<()> = flow.cluster::<()>();
37 /// let numbers: Stream<_, Process<_>, _> = p1.source_iter(q!(vec![0, 1, 2, 3]));
38 /// let on_worker: Stream<_, Cluster<_>, _> = numbers
39 /// .map(q!(|x| (hydro_lang::location::MemberId::from_raw_id(x), x)))
40 /// .into_keyed()
41 /// .demux_bincode(&workers);
42 /// # on_worker.send_bincode(&p2).entries()
43 /// // if there are 4 members in the cluster, each receives one element
44 /// // - MemberId::<()>(0): [0]
45 /// // - MemberId::<()>(1): [1]
46 /// // - MemberId::<()>(2): [2]
47 /// // - MemberId::<()>(3): [3]
48 /// # }, |mut stream| async move {
49 /// # let mut results = Vec::new();
50 /// # for w in 0..4 {
51 /// # results.push(format!("{:?}", stream.next().await.unwrap()));
52 /// # }
53 /// # results.sort();
54 /// # assert_eq!(results, vec!["(MemberId::<()>(0), 0)", "(MemberId::<()>(1), 1)", "(MemberId::<()>(2), 2)", "(MemberId::<()>(3), 3)"]);
55 /// # }));
56 /// # }
57 /// ```
58 pub fn demux_bincode(
59 self,
60 other: &Cluster<'a, L2>,
61 ) -> Stream<T, Cluster<'a, L2>, Unbounded, O, R>
62 where
63 T: Serialize + DeserializeOwned,
64 {
65 self.demux(other, TCP.fail_stop().bincode())
66 }
67
68 /// Sends each group of this stream to a specific member of a cluster, with the [`MemberId`] key
69 /// identifying the recipient for each group and using the configuration in `via` to set up the
70 /// message transport.
71 ///
72 /// Each key must be a `MemberId<L2>` and each value must be a `T` where the key specifies
73 /// which cluster member should receive the data. Unlike [`Stream::broadcast`], this
74 /// API allows precise targeting of specific cluster members rather than broadcasting to
75 /// all members.
76 ///
77 /// # Example
78 /// ```rust
79 /// # #[cfg(feature = "deploy")] {
80 /// # use hydro_lang::prelude::*;
81 /// # use futures::StreamExt;
82 /// # tokio_test::block_on(hydro_lang::test_util::multi_location_test(|flow, p2| {
83 /// let p1 = flow.process::<()>();
84 /// let workers: Cluster<()> = flow.cluster::<()>();
85 /// let numbers: Stream<_, Process<_>, _> = p1.source_iter(q!(vec![0, 1, 2, 3]));
86 /// let on_worker: Stream<_, Cluster<_>, _> = numbers
87 /// .map(q!(|x| (hydro_lang::location::MemberId::from_raw_id(x), x)))
88 /// .into_keyed()
89 /// .demux(&workers, TCP.fail_stop().bincode());
90 /// # on_worker.send(&p2, TCP.fail_stop().bincode()).entries()
91 /// // if there are 4 members in the cluster, each receives one element
92 /// // - MemberId::<()>(0): [0]
93 /// // - MemberId::<()>(1): [1]
94 /// // - MemberId::<()>(2): [2]
95 /// // - MemberId::<()>(3): [3]
96 /// # }, |mut stream| async move {
97 /// # let mut results = Vec::new();
98 /// # for w in 0..4 {
99 /// # results.push(format!("{:?}", stream.next().await.unwrap()));
100 /// # }
101 /// # results.sort();
102 /// # assert_eq!(results, vec!["(MemberId::<()>(0), 0)", "(MemberId::<()>(1), 1)", "(MemberId::<()>(2), 2)", "(MemberId::<()>(3), 3)"]);
103 /// # }));
104 /// # }
105 /// ```
106 pub fn demux<N: NetworkFor<T>>(
107 self,
108 to: &Cluster<'a, L2>,
109 via: N,
110 ) -> Stream<
111 T,
112 // NoConsistency because there each replica member may receive different streams
113 Cluster<'a, L2, NoConsistency>,
114 Unbounded,
115 <O as MinOrder<N::OrderingGuarantee>>::Min,
116 R,
117 >
118 where
119 T: Serialize + DeserializeOwned,
120 O: MinOrder<N::OrderingGuarantee>,
121 {
122 let name = via.name();
123 if to.multiversioned() && name.is_none() {
124 panic!(
125 "Cannot send to a multiversioned location without a channel name. Please provide a name for the network."
126 );
127 }
128
129 let (serialize, deserialize) = if N::is_embedded() {
130 (
131 NetworkSend::Embedded {
132 tag: Some(quote_type::<L2>().into()),
133 element_type: quote_type::<T>().into(),
134 },
135 NetworkRecv::Embedded {
136 tag: None,
137 element_type: quote_type::<T>().into(),
138 },
139 )
140 } else {
141 (
142 NetworkSend::Custom {
143 serialize_fn: Some(N::serialize_thunk(true).into()),
144 },
145 NetworkRecv::Custom {
146 deserialize_fn: Some(N::deserialize_thunk(None).into()),
147 },
148 )
149 };
150
151 Stream::new(
152 to.clone(),
153 HydroNode::Network {
154 name: name.map(ToOwned::to_owned),
155 networking_info: N::networking_info(),
156 serialize,
157 deserialize,
158 instantiate_fn: DebugInstantiate::Building,
159 input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
160 metadata: to.new_node_metadata(Stream::<
161 T,
162 Cluster<'a, L2>,
163 Unbounded,
164 <O as MinOrder<N::OrderingGuarantee>>::Min,
165 R,
166 >::collection_kind()),
167 },
168 )
169 }
170}
171
172impl<'a, K, T, L, L2, B: Boundedness, O: Ordering, R: Retries>
173 KeyedStream<(MemberId<L2>, K), T, Process<'a, L>, B, O, R>
174{
175 #[deprecated = "use KeyedStream::demux(..., TCP.fail_stop().bincode()) instead"]
176 /// Sends each group of this stream to a specific member of a cluster. The input stream has a
177 /// compound key where the first element is the recipient's [`MemberId`] and the second element
178 /// is a key that will be sent along with the value, using [`bincode`] to serialize/deserialize
179 /// messages.
180 ///
181 /// # Example
182 /// ```rust
183 /// # #[cfg(feature = "deploy")] {
184 /// # use hydro_lang::prelude::*;
185 /// # use futures::StreamExt;
186 /// # tokio_test::block_on(hydro_lang::test_util::multi_location_test(|flow, p2| {
187 /// let p1 = flow.process::<()>();
188 /// let workers: Cluster<()> = flow.cluster::<()>();
189 /// let to_send: KeyedStream<_, _, Process<_>, _> = p1
190 /// .source_iter(q!(vec![0, 1, 2, 3]))
191 /// .map(q!(|x| ((hydro_lang::location::MemberId::from_raw_id(x), x), x + 123)))
192 /// .into_keyed();
193 /// let on_worker: KeyedStream<_, _, Cluster<_>, _> = to_send.demux_bincode(&workers);
194 /// # on_worker.entries().send_bincode(&p2).entries()
195 /// // if there are 4 members in the cluster, each receives one element
196 /// // - MemberId::<()>(0): { 0: [123] }
197 /// // - MemberId::<()>(1): { 1: [124] }
198 /// // - ...
199 /// # }, |mut stream| async move {
200 /// # let mut results = Vec::new();
201 /// # for w in 0..4 {
202 /// # results.push(format!("{:?}", stream.next().await.unwrap()));
203 /// # }
204 /// # results.sort();
205 /// # assert_eq!(results, vec!["(MemberId::<()>(0), (0, 123))", "(MemberId::<()>(1), (1, 124))", "(MemberId::<()>(2), (2, 125))", "(MemberId::<()>(3), (3, 126))"]);
206 /// # }));
207 /// # }
208 /// ```
209 pub fn demux_bincode(
210 self,
211 other: &Cluster<'a, L2>,
212 ) -> KeyedStream<K, T, Cluster<'a, L2>, Unbounded, O, R>
213 where
214 K: Serialize + DeserializeOwned,
215 T: Serialize + DeserializeOwned,
216 {
217 self.demux(other, TCP.fail_stop().bincode())
218 }
219
220 /// Sends each group of this stream to a specific member of a cluster. The input stream has a
221 /// compound key where the first element is the recipient's [`MemberId`] and the second element
222 /// is a key that will be sent along with the value, using the configuration in `via` to set up
223 /// the message transport.
224 ///
225 /// # Example
226 /// ```rust
227 /// # #[cfg(feature = "deploy")] {
228 /// # use hydro_lang::prelude::*;
229 /// # use futures::StreamExt;
230 /// # tokio_test::block_on(hydro_lang::test_util::multi_location_test(|flow, p2| {
231 /// let p1 = flow.process::<()>();
232 /// let workers: Cluster<()> = flow.cluster::<()>();
233 /// let to_send: KeyedStream<_, _, Process<_>, _> = p1
234 /// .source_iter(q!(vec![0, 1, 2, 3]))
235 /// .map(q!(|x| ((hydro_lang::location::MemberId::from_raw_id(x), x), x + 123)))
236 /// .into_keyed();
237 /// let on_worker: KeyedStream<_, _, Cluster<_>, _> = to_send.demux(&workers, TCP.fail_stop().bincode());
238 /// # on_worker.entries().send(&p2, TCP.fail_stop().bincode()).entries()
239 /// // if there are 4 members in the cluster, each receives one element
240 /// // - MemberId::<()>(0): { 0: [123] }
241 /// // - MemberId::<()>(1): { 1: [124] }
242 /// // - ...
243 /// # }, |mut stream| async move {
244 /// # let mut results = Vec::new();
245 /// # for w in 0..4 {
246 /// # results.push(format!("{:?}", stream.next().await.unwrap()));
247 /// # }
248 /// # results.sort();
249 /// # assert_eq!(results, vec!["(MemberId::<()>(0), (0, 123))", "(MemberId::<()>(1), (1, 124))", "(MemberId::<()>(2), (2, 125))", "(MemberId::<()>(3), (3, 126))"]);
250 /// # }));
251 /// # }
252 /// ```
253 pub fn demux<N: NetworkFor<(K, T)>>(
254 self,
255 to: &Cluster<'a, L2>,
256 via: N,
257 ) -> KeyedStream<
258 K,
259 T,
260 Cluster<'a, L2, NoConsistency>,
261 Unbounded,
262 <O as MinOrder<N::OrderingGuarantee>>::Min,
263 R,
264 >
265 where
266 K: Serialize + DeserializeOwned,
267 T: Serialize + DeserializeOwned,
268 O: MinOrder<N::OrderingGuarantee>,
269 {
270 let name = via.name();
271 if to.multiversioned() && name.is_none() {
272 panic!(
273 "Cannot send to a multiversioned location without a channel name. Please provide a name for the network."
274 );
275 }
276
277 let (serialize, deserialize) = if N::is_embedded() {
278 (
279 NetworkSend::Embedded {
280 tag: Some(quote_type::<L2>().into()),
281 element_type: quote_type::<(K, T)>().into(),
282 },
283 NetworkRecv::Embedded {
284 tag: None,
285 element_type: quote_type::<(K, T)>().into(),
286 },
287 )
288 } else {
289 (
290 NetworkSend::Custom {
291 serialize_fn: Some(N::serialize_thunk(true).into()),
292 },
293 NetworkRecv::Custom {
294 deserialize_fn: Some(N::deserialize_thunk(None).into()),
295 },
296 )
297 };
298
299 KeyedStream::new(
300 to.clone(),
301 HydroNode::Network {
302 name: name.map(ToOwned::to_owned),
303 networking_info: N::networking_info(),
304 serialize,
305 deserialize,
306 instantiate_fn: DebugInstantiate::Building,
307 input: Box::new(
308 self.entries()
309 .map(q!(|((id, k), v)| (id, (k, v))))
310 .ir_node
311 .replace(HydroNode::Placeholder),
312 ),
313 metadata: to.new_node_metadata(KeyedStream::<
314 K,
315 T,
316 Cluster<'a, L2>,
317 Unbounded,
318 <O as MinOrder<N::OrderingGuarantee>>::Min,
319 R,
320 >::collection_kind()),
321 },
322 )
323 }
324}
325
326impl<'a, T, L, L2, B: Boundedness, C: Consistency, O: Ordering, R: Retries>
327 KeyedStream<MemberId<L2>, T, Cluster<'a, L, C>, B, O, R>
328{
329 #[deprecated = "use KeyedStream::demux(..., TCP.fail_stop().bincode()) instead"]
330 /// Sends each group of this stream at each source member to a specific member of a destination
331 /// cluster, with the [`MemberId`] key identifying the recipient for each group and using
332 /// [`bincode`] to serialize/deserialize messages.
333 ///
334 /// Each key must be a `MemberId<L2>` and each value must be a `T` where the key specifies
335 /// which cluster member should receive the data. Unlike [`Stream::broadcast_bincode`], this
336 /// API allows precise targeting of specific cluster members rather than broadcasting to all
337 /// members.
338 ///
339 /// Each cluster member sends its local stream elements, and they are collected at each
340 /// destination member as a [`KeyedStream`] where keys identify the source cluster member.
341 ///
342 /// # Example
343 /// ```rust
344 /// # #[cfg(feature = "deploy")] {
345 /// # use hydro_lang::prelude::*;
346 /// # use futures::StreamExt;
347 /// # tokio_test::block_on(hydro_lang::test_util::multi_location_test(|flow, p2| {
348 /// # type Source = ();
349 /// # type Destination = ();
350 /// let source: Cluster<Source> = flow.cluster::<Source>();
351 /// let to_send: KeyedStream<_, _, Cluster<_>, _> = source
352 /// .source_iter(q!(vec![0, 1, 2, 3]))
353 /// .map(q!(|x| (hydro_lang::location::MemberId::from_raw_id(x), x)))
354 /// .into_keyed();
355 /// let destination: Cluster<Destination> = flow.cluster::<Destination>();
356 /// let all_received = to_send.demux_bincode(&destination); // KeyedStream<MemberId<Source>, i32, ...>
357 /// # all_received.entries().send_bincode(&p2).entries()
358 /// # }, |mut stream| async move {
359 /// // if there are 4 members in the destination cluster, each receives one message from each source member
360 /// // - Destination(0): { Source(0): [0], Source(1): [0], ... }
361 /// // - Destination(1): { Source(0): [1], Source(1): [1], ... }
362 /// // - ...
363 /// # let mut results = Vec::new();
364 /// # for w in 0..16 {
365 /// # results.push(format!("{:?}", stream.next().await.unwrap()));
366 /// # }
367 /// # results.sort();
368 /// # assert_eq!(results, vec![
369 /// # "(MemberId::<()>(0), (MemberId::<()>(0), 0))", "(MemberId::<()>(0), (MemberId::<()>(1), 0))", "(MemberId::<()>(0), (MemberId::<()>(2), 0))", "(MemberId::<()>(0), (MemberId::<()>(3), 0))",
370 /// # "(MemberId::<()>(1), (MemberId::<()>(0), 1))", "(MemberId::<()>(1), (MemberId::<()>(1), 1))", "(MemberId::<()>(1), (MemberId::<()>(2), 1))", "(MemberId::<()>(1), (MemberId::<()>(3), 1))",
371 /// # "(MemberId::<()>(2), (MemberId::<()>(0), 2))", "(MemberId::<()>(2), (MemberId::<()>(1), 2))", "(MemberId::<()>(2), (MemberId::<()>(2), 2))", "(MemberId::<()>(2), (MemberId::<()>(3), 2))",
372 /// # "(MemberId::<()>(3), (MemberId::<()>(0), 3))", "(MemberId::<()>(3), (MemberId::<()>(1), 3))", "(MemberId::<()>(3), (MemberId::<()>(2), 3))", "(MemberId::<()>(3), (MemberId::<()>(3), 3))"
373 /// # ]);
374 /// # }));
375 /// # }
376 /// ```
377 pub fn demux_bincode(
378 self,
379 other: &Cluster<'a, L2>,
380 ) -> KeyedStream<MemberId<L>, T, Cluster<'a, L2>, Unbounded, O, R>
381 where
382 T: Serialize + DeserializeOwned,
383 {
384 self.demux(other, TCP.fail_stop().bincode())
385 }
386
387 /// Sends each group of this stream at each source member to a specific member of a destination
388 /// cluster, with the [`MemberId`] key identifying the recipient for each group and using the
389 /// configuration in `via` to set up the message transport.
390 ///
391 /// Each key must be a `MemberId<L2>` and each value must be a `T` where the key specifies
392 /// which cluster member should receive the data. Unlike [`Stream::broadcast`], this
393 /// API allows precise targeting of specific cluster members rather than broadcasting to all
394 /// members.
395 ///
396 /// Each cluster member sends its local stream elements, and they are collected at each
397 /// destination member as a [`KeyedStream`] where keys identify the source cluster member.
398 ///
399 /// # Example
400 /// ```rust
401 /// # #[cfg(feature = "deploy")] {
402 /// # use hydro_lang::prelude::*;
403 /// # use futures::StreamExt;
404 /// # tokio_test::block_on(hydro_lang::test_util::multi_location_test(|flow, p2| {
405 /// # type Source = ();
406 /// # type Destination = ();
407 /// let source: Cluster<Source> = flow.cluster::<Source>();
408 /// let to_send: KeyedStream<_, _, Cluster<_>, _> = source
409 /// .source_iter(q!(vec![0, 1, 2, 3]))
410 /// .map(q!(|x| (hydro_lang::location::MemberId::from_raw_id(x), x)))
411 /// .into_keyed();
412 /// let destination: Cluster<Destination> = flow.cluster::<Destination>();
413 /// let all_received = to_send.demux(&destination, TCP.fail_stop().bincode()); // KeyedStream<MemberId<Source>, i32, ...>
414 /// # all_received.entries().send(&p2, TCP.fail_stop().bincode()).entries()
415 /// # }, |mut stream| async move {
416 /// // if there are 4 members in the destination cluster, each receives one message from each source member
417 /// // - Destination(0): { Source(0): [0], Source(1): [0], ... }
418 /// // - Destination(1): { Source(0): [1], Source(1): [1], ... }
419 /// // - ...
420 /// # let mut results = Vec::new();
421 /// # for w in 0..16 {
422 /// # results.push(format!("{:?}", stream.next().await.unwrap()));
423 /// # }
424 /// # results.sort();
425 /// # assert_eq!(results, vec![
426 /// # "(MemberId::<()>(0), (MemberId::<()>(0), 0))", "(MemberId::<()>(0), (MemberId::<()>(1), 0))", "(MemberId::<()>(0), (MemberId::<()>(2), 0))", "(MemberId::<()>(0), (MemberId::<()>(3), 0))",
427 /// # "(MemberId::<()>(1), (MemberId::<()>(0), 1))", "(MemberId::<()>(1), (MemberId::<()>(1), 1))", "(MemberId::<()>(1), (MemberId::<()>(2), 1))", "(MemberId::<()>(1), (MemberId::<()>(3), 1))",
428 /// # "(MemberId::<()>(2), (MemberId::<()>(0), 2))", "(MemberId::<()>(2), (MemberId::<()>(1), 2))", "(MemberId::<()>(2), (MemberId::<()>(2), 2))", "(MemberId::<()>(2), (MemberId::<()>(3), 2))",
429 /// # "(MemberId::<()>(3), (MemberId::<()>(0), 3))", "(MemberId::<()>(3), (MemberId::<()>(1), 3))", "(MemberId::<()>(3), (MemberId::<()>(2), 3))", "(MemberId::<()>(3), (MemberId::<()>(3), 3))"
430 /// # ]);
431 /// # }));
432 /// # }
433 /// ```
434 pub fn demux<N: NetworkFor<T>>(
435 self,
436 to: &Cluster<'a, L2>,
437 via: N,
438 ) -> KeyedStream<
439 MemberId<L>,
440 T,
441 Cluster<'a, L2, NoConsistency>,
442 Unbounded,
443 <O as MinOrder<N::OrderingGuarantee>>::Min,
444 R,
445 >
446 where
447 T: Serialize + DeserializeOwned,
448 O: MinOrder<N::OrderingGuarantee>,
449 {
450 let name = via.name();
451 if to.multiversioned() && name.is_none() {
452 panic!(
453 "Cannot send to a multiversioned location without a channel name. Please provide a name for the network."
454 );
455 }
456
457 let (serialize, deserialize) = if N::is_embedded() {
458 (
459 NetworkSend::Embedded {
460 tag: Some(quote_type::<L2>().into()),
461 element_type: quote_type::<T>().into(),
462 },
463 NetworkRecv::Embedded {
464 tag: Some(quote_type::<L>().into()),
465 element_type: quote_type::<T>().into(),
466 },
467 )
468 } else {
469 (
470 NetworkSend::Custom {
471 serialize_fn: Some(N::serialize_thunk(true).into()),
472 },
473 NetworkRecv::Custom {
474 deserialize_fn: Some(N::deserialize_thunk(Some("e_type::<L>())).into()),
475 },
476 )
477 };
478
479 KeyedStream::new(
480 to.clone(),
481 HydroNode::Network {
482 name: name.map(ToOwned::to_owned),
483 networking_info: N::networking_info(),
484 serialize,
485 deserialize,
486 instantiate_fn: DebugInstantiate::Building,
487 input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
488 metadata: to.new_node_metadata(KeyedStream::<
489 MemberId<L>,
490 T,
491 Cluster<'a, L2>,
492 Unbounded,
493 <O as MinOrder<N::OrderingGuarantee>>::Min,
494 R,
495 >::collection_kind()),
496 },
497 )
498 }
499}
500
501impl<'a, K, V, L, B: Boundedness, C: Consistency, O: Ordering, R: Retries>
502 KeyedStream<K, V, Cluster<'a, L, C>, B, O, R>
503{
504 #[deprecated = "use KeyedStream::send(..., TCP.fail_stop().bincode()) instead"]
505 /// "Moves" elements of this keyed stream from a cluster to a process by sending them over the
506 /// network, using [`bincode`] to serialize/deserialize messages. The resulting [`KeyedStream`]
507 /// has a compound key where the first element is the sender's [`MemberId`] and the second
508 /// element is the original key.
509 ///
510 /// # Example
511 /// ```rust
512 /// # #[cfg(feature = "deploy")] {
513 /// # use hydro_lang::prelude::*;
514 /// # use futures::StreamExt;
515 /// # tokio_test::block_on(hydro_lang::test_util::multi_location_test(|flow, p2| {
516 /// # type Source = ();
517 /// # type Destination = ();
518 /// let source: Cluster<Source> = flow.cluster::<Source>();
519 /// let to_send: KeyedStream<_, _, Cluster<_>, _> = source
520 /// .source_iter(q!(vec![0, 1, 2, 3]))
521 /// .map(q!(|x| (x, x + 123)))
522 /// .into_keyed();
523 /// let destination_process = flow.process::<Destination>();
524 /// let all_received = to_send.send_bincode(&destination_process); // KeyedStream<(MemberId<Source>, i32), i32, ...>
525 /// # all_received.entries().send_bincode(&p2)
526 /// # }, |mut stream| async move {
527 /// // if there are 4 members in the source cluster, the destination process receives four messages from each source member
528 /// // {
529 /// // (MemberId<Source>(0), 0): [123], (MemberId<Source>(1), 0): [123], ...,
530 /// // (MemberId<Source>(0), 1): [124], (MemberId<Source>(1), 1): [124], ...,
531 /// // ...
532 /// // }
533 /// # let mut results = Vec::new();
534 /// # for w in 0..16 {
535 /// # results.push(format!("{:?}", stream.next().await.unwrap()));
536 /// # }
537 /// # results.sort();
538 /// # assert_eq!(results, vec![
539 /// # "((MemberId::<()>(0), 0), 123)",
540 /// # "((MemberId::<()>(0), 1), 124)",
541 /// # "((MemberId::<()>(0), 2), 125)",
542 /// # "((MemberId::<()>(0), 3), 126)",
543 /// # "((MemberId::<()>(1), 0), 123)",
544 /// # "((MemberId::<()>(1), 1), 124)",
545 /// # "((MemberId::<()>(1), 2), 125)",
546 /// # "((MemberId::<()>(1), 3), 126)",
547 /// # "((MemberId::<()>(2), 0), 123)",
548 /// # "((MemberId::<()>(2), 1), 124)",
549 /// # "((MemberId::<()>(2), 2), 125)",
550 /// # "((MemberId::<()>(2), 3), 126)",
551 /// # "((MemberId::<()>(3), 0), 123)",
552 /// # "((MemberId::<()>(3), 1), 124)",
553 /// # "((MemberId::<()>(3), 2), 125)",
554 /// # "((MemberId::<()>(3), 3), 126)",
555 /// # ]);
556 /// # }));
557 /// # }
558 /// ```
559 pub fn send_bincode<L2>(
560 self,
561 other: &Process<'a, L2>,
562 ) -> KeyedStream<(MemberId<L>, K), V, Process<'a, L2>, Unbounded, O, R>
563 where
564 K: Serialize + DeserializeOwned,
565 V: Serialize + DeserializeOwned,
566 {
567 self.send(other, TCP.fail_stop().bincode())
568 }
569
570 /// "Moves" elements of this keyed stream from a cluster to a process by sending them over the
571 /// network, using the configuration in `via` to set up the message transport. The resulting
572 /// [`KeyedStream`] has a compound key where the first element is the sender's [`MemberId`] and
573 /// the second element is the original key.
574 ///
575 /// # Example
576 /// ```rust
577 /// # #[cfg(feature = "deploy")] {
578 /// # use hydro_lang::prelude::*;
579 /// # use futures::StreamExt;
580 /// # tokio_test::block_on(hydro_lang::test_util::multi_location_test(|flow, p2| {
581 /// # type Source = ();
582 /// # type Destination = ();
583 /// let source: Cluster<Source> = flow.cluster::<Source>();
584 /// let to_send: KeyedStream<_, _, Cluster<_>, _> = source
585 /// .source_iter(q!(vec![0, 1, 2, 3]))
586 /// .map(q!(|x| (x, x + 123)))
587 /// .into_keyed();
588 /// let destination_process = flow.process::<Destination>();
589 /// let all_received = to_send.send(&destination_process, TCP.fail_stop().bincode()); // KeyedStream<(MemberId<Source>, i32), i32, ...>
590 /// # all_received.entries().send(&p2, TCP.fail_stop().bincode())
591 /// # }, |mut stream| async move {
592 /// // if there are 4 members in the source cluster, the destination process receives four messages from each source member
593 /// // {
594 /// // (MemberId<Source>(0), 0): [123], (MemberId<Source>(1), 0): [123], ...,
595 /// // (MemberId<Source>(0), 1): [124], (MemberId<Source>(1), 1): [124], ...,
596 /// // ...
597 /// // }
598 /// # let mut results = Vec::new();
599 /// # for w in 0..16 {
600 /// # results.push(format!("{:?}", stream.next().await.unwrap()));
601 /// # }
602 /// # results.sort();
603 /// # assert_eq!(results, vec![
604 /// # "((MemberId::<()>(0), 0), 123)",
605 /// # "((MemberId::<()>(0), 1), 124)",
606 /// # "((MemberId::<()>(0), 2), 125)",
607 /// # "((MemberId::<()>(0), 3), 126)",
608 /// # "((MemberId::<()>(1), 0), 123)",
609 /// # "((MemberId::<()>(1), 1), 124)",
610 /// # "((MemberId::<()>(1), 2), 125)",
611 /// # "((MemberId::<()>(1), 3), 126)",
612 /// # "((MemberId::<()>(2), 0), 123)",
613 /// # "((MemberId::<()>(2), 1), 124)",
614 /// # "((MemberId::<()>(2), 2), 125)",
615 /// # "((MemberId::<()>(2), 3), 126)",
616 /// # "((MemberId::<()>(3), 0), 123)",
617 /// # "((MemberId::<()>(3), 1), 124)",
618 /// # "((MemberId::<()>(3), 2), 125)",
619 /// # "((MemberId::<()>(3), 3), 126)",
620 /// # ]);
621 /// # }));
622 /// # }
623 /// ```
624 pub fn send<L2, N: NetworkFor<(K, V)>>(
625 self,
626 to: &Process<'a, L2>,
627 via: N,
628 ) -> KeyedStream<
629 (MemberId<L>, K),
630 V,
631 Process<'a, L2>,
632 Unbounded,
633 <O as MinOrder<N::OrderingGuarantee>>::Min,
634 R,
635 >
636 where
637 K: Serialize + DeserializeOwned,
638 V: Serialize + DeserializeOwned,
639 O: MinOrder<N::OrderingGuarantee>,
640 {
641 let name = via.name();
642 if to.multiversioned() && name.is_none() {
643 panic!(
644 "Cannot send to a multiversioned location without a channel name. Please provide a name for the network."
645 );
646 }
647
648 let (serialize, deserialize) = if N::is_embedded() {
649 (
650 NetworkSend::Embedded {
651 tag: None,
652 element_type: quote_type::<(K, V)>().into(),
653 },
654 NetworkRecv::Embedded {
655 tag: Some(quote_type::<L>().into()),
656 element_type: quote_type::<(K, V)>().into(),
657 },
658 )
659 } else {
660 (
661 NetworkSend::Custom {
662 serialize_fn: Some(N::serialize_thunk(false).into()),
663 },
664 NetworkRecv::Custom {
665 deserialize_fn: Some(N::deserialize_thunk(Some("e_type::<L>())).into()),
666 },
667 )
668 };
669
670 let raw_stream: Stream<
671 (MemberId<L>, (K, V)),
672 Process<'a, L2>,
673 Unbounded,
674 <O as MinOrder<N::OrderingGuarantee>>::Min,
675 R,
676 > = Stream::new(
677 to.clone(),
678 HydroNode::Network {
679 name: name.map(ToOwned::to_owned),
680 networking_info: N::networking_info(),
681 serialize,
682 deserialize,
683 instantiate_fn: DebugInstantiate::Building,
684 input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
685 metadata: to.new_node_metadata(Stream::<
686 (MemberId<L>, (K, V)),
687 Cluster<'a, L2>,
688 Unbounded,
689 <O as MinOrder<N::OrderingGuarantee>>::Min,
690 R,
691 >::collection_kind()),
692 },
693 );
694
695 raw_stream
696 .map(q!(|(sender, (k, v))| ((sender, k), v)))
697 .into_keyed()
698 }
699}