1 // Copyright 2022 gRPC authors.
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License");
4 // you may not use this file except in compliance with the License.
5 // You may obtain a copy of the License at
6 //
7 // http://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS,
11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 // See the License for the specific language governing permissions and
13 // limitations under the License.
14
15 #ifndef GRPC_SRC_CORE_LIB_PROMISE_MAP_PIPE_H
16 #define GRPC_SRC_CORE_LIB_PROMISE_MAP_PIPE_H
17
18 #include <grpc/support/port_platform.h>
19
20 #include "absl/status/status.h"
21
22 #include <grpc/support/log.h>
23
24 #include "src/core/lib/promise/detail/promise_factory.h"
25 #include "src/core/lib/promise/for_each.h"
26 #include "src/core/lib/promise/map.h"
27 #include "src/core/lib/promise/pipe.h"
28 #include "src/core/lib/promise/poll.h"
29 #include "src/core/lib/promise/trace.h"
30 #include "src/core/lib/promise/try_seq.h"
31
32 namespace grpc_core {
33
34 // Apply a (possibly async) mapping function to src, and output into dst.
35 //
36 // In psuedo-code:
37 // for each element in wait_for src.Next:
38 // x = wait_for filter_factory(element)
39 // wait_for dst.Push(x)
40 template <typename T, typename Filter>
MapPipe(PipeReceiver<T> src,PipeSender<T> dst,Filter filter_factory)41 auto MapPipe(PipeReceiver<T> src, PipeSender<T> dst, Filter filter_factory) {
42 return ForEach(
43 std::move(src),
44 [filter_factory = promise_detail::RepeatedPromiseFactory<T, Filter>(
45 std::move(filter_factory)),
46 dst = std::move(dst)](T t) mutable {
47 return TrySeq(
48 [] {
49 if (grpc_trace_promise_primitives.enabled()) {
50 gpr_log(GPR_DEBUG, "MapPipe: start map");
51 }
52 return Empty{};
53 },
54 filter_factory.Make(std::move(t)),
55 [&dst](T t) {
56 if (grpc_trace_promise_primitives.enabled()) {
57 gpr_log(GPR_DEBUG, "MapPipe: start push");
58 }
59 return Map(dst.Push(std::move(t)), [](bool successful_push) {
60 if (successful_push) {
61 return absl::OkStatus();
62 }
63 return absl::CancelledError();
64 });
65 });
66 });
67 }
68
69 // Helper to intecept a pipe and apply a mapping function.
70 // Each of the `Intercept` constructors will take a PipeSender or PipeReceiver,
71 // construct a new pipe, and then replace the passed in pipe with its new end.
72 // In this way it can interject logic per-element.
73 // Next, the TakeAndRun function will return a promise that can be run to apply
74 // a mapping promise to each element of the pipe.
75 template <typename T>
76 class PipeMapper {
77 public:
Intercept(PipeSender<T> & intercept_sender)78 static PipeMapper Intercept(PipeSender<T>& intercept_sender) {
79 PipeMapper<T> r;
80 r.interceptor_.sender.Swap(&intercept_sender);
81 return r;
82 }
83
Intercept(PipeReceiver<T> & intercept_receiver)84 static PipeMapper Intercept(PipeReceiver<T>& intercept_receiver) {
85 PipeMapper<T> r;
86 r.interceptor_.receiver.Swap(&intercept_receiver);
87 return r;
88 }
89
90 template <typename Filter>
TakeAndRun(Filter filter)91 auto TakeAndRun(Filter filter) {
92 return MapPipe(std::move(interceptor_.receiver),
93 std::move(interceptor_.sender), std::move(filter));
94 }
95
96 private:
97 PipeMapper() = default;
98 Pipe<T> interceptor_;
99 };
100
101 } // namespace grpc_core
102
103 #endif // GRPC_SRC_CORE_LIB_PROMISE_MAP_PIPE_H
104