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