| 2103 | } |
| 2104 | |
| 2105 | func mapExecutorStreamChunks(ctx context.Context, in <-chan pluginapi.ExecutorStreamChunk) <-chan coreexecutor.StreamChunk { |
| 2106 | if ctx == nil { |
| 2107 | ctx = context.Background() |
| 2108 | } |
| 2109 | out := make(chan coreexecutor.StreamChunk) |
| 2110 | if in == nil { |
| 2111 | close(out) |
| 2112 | return out |
| 2113 | } |
| 2114 | go func() { |
| 2115 | defer close(out) |
| 2116 | for { |
| 2117 | var mapped coreexecutor.StreamChunk |
| 2118 | select { |
| 2119 | case <-ctx.Done(): |
| 2120 | return |
| 2121 | case chunk, ok := <-in: |
| 2122 | if !ok { |
| 2123 | return |
| 2124 | } |
| 2125 | mapped = coreexecutor.StreamChunk{ |
| 2126 | Payload: bytes.Clone(chunk.Payload), |
| 2127 | Err: chunk.Err, |
| 2128 | } |
| 2129 | } |
| 2130 | select { |
| 2131 | case <-ctx.Done(): |
| 2132 | return |
| 2133 | case out <- mapped: |
| 2134 | } |
| 2135 | } |
| 2136 | }() |
| 2137 | return out |
| 2138 | } |
| 2139 | |
| 2140 | func readAndRestoreRequestBody(r *http.Request) ([]byte, error) { |
| 2141 | if r == nil || r.Body == nil { |