Method handleDataStream

connection/quic_connection.go:189–216

(ctx context.Context, stream *rpcquic.RequestServerStream)

}

func (q *quicConnection) handleDataStream(ctx context.Context, stream *rpcquic.RequestServerStream) error {
	request, err := stream.ReadConnectRequestData()
	if err != nil {
		return err
	}

	if err, connectResponseSent := q.dispatchRequest(ctx, stream, request); err != nil {
		q.logger.Err(err).Str("type", request.Type.String()).Str("dest", request.Dest).Msg("Request failed")

		// if the connectResponse was already sent and we had an error, we need to propagate it up, so that the stream is
		// closed with an RST_STREAM frame
		if connectResponseSent {
			return err
		}

		var metadata []pogs.Metadata
		// Check the type of error that was throw and add metadata that will help identify it on OTD.
		if errors.Is(err, cfdflow.ErrTooManyActiveFlows) {
			metadata = append(metadata, pogs.ErrorFlowConnectRateLimitedMetadata)
		}

		if writeRespErr := stream.WriteConnectResponseData(err, metadata...); writeRespErr != nil {
			return writeRespErr
		}
	}

	return nil
}

// dispatchRequest will dispatch the request to the origin depending on the type and returns an error if it occurs.
// Also returns if the connect response was sent to the downstream during processing of the origin request.