(ctx context.Context, req pluginapi.StreamChunkInterceptRequest, skipPluginID string)
| 653 | } |
| 654 | |
| 655 | func (h *Host) InterceptStreamChunkExcept(ctx context.Context, req pluginapi.StreamChunkInterceptRequest, skipPluginID string) pluginapi.StreamChunkInterceptResponse { |
| 656 | current := pluginapi.StreamChunkInterceptResponse{ |
| 657 | Headers: cloneHeader(req.ResponseHeaders), |
| 658 | Body: bytes.Clone(req.Body), |
| 659 | } |
| 660 | skipPluginID = strings.TrimSpace(skipPluginID) |
| 661 | for _, record := range h.activeRecords() { |
| 662 | interceptor := record.plugin.Capabilities.StreamChunkInterceptor |
| 663 | if h.isPluginFused(record.id) || interceptor == nil || current.DropChunk || record.id == skipPluginID { |
| 664 | continue |
| 665 | } |
| 666 | nextReq := req |
| 667 | nextReq.RequestHeaders = cloneHeader(req.RequestHeaders) |
| 668 | nextReq.ResponseHeaders = cloneHeader(current.Headers) |
| 669 | nextReq.OriginalRequest = bytes.Clone(req.OriginalRequest) |
| 670 | nextReq.RequestBody = bytes.Clone(req.RequestBody) |
| 671 | nextReq.Body = bytes.Clone(current.Body) |
| 672 | nextReq.HistoryChunks = cloneByteSlices(req.HistoryChunks) |
| 673 | nextReq.Metadata = cloneInterceptorMetadata(req.Metadata) |
| 674 | if resp, ok := h.callStreamChunkInterceptor(ctx, record, interceptor, nextReq); ok { |
| 675 | current.Headers = mergeHeaders(current.Headers, resp.Headers, resp.ClearHeaders) |
| 676 | if len(resp.Body) > 0 { |
| 677 | current.Body = bytes.Clone(resp.Body) |
| 678 | } |
| 679 | if resp.DropChunk { |
| 680 | current.DropChunk = true |
| 681 | } |
| 682 | } |
| 683 | } |
| 684 | return current |
| 685 | } |
| 686 | |
| 687 | func (h *Host) HasStreamInterceptors() bool { |
| 688 | if h == nil { |
no test coverage detected