When using text/event-stream content type, the results are not refreshed in real time, but returned all at once

[alivy]

Steps to Reproduce the Problem

1. 当响应体中有text/event-stream时，返回结果为一次性加载返回，而并非实时刷新
2. 即使我加载了自己的Middleware添加到OcelotPipelineExtensions时，也不能实时刷新，以异步流的形式收到结果
   自定义拓展代码如下

    public class CustomResponseMiddleware : OcelotMiddleware
    {
        private readonly RequestDelegate _next;
        public CustomResponseMiddleware(RequestDelegate next, IOcelotLoggerFactory loggerFactory)
            : base(loggerFactory.CreateLogger<CustomResponseMiddleware>())
        {
            _next = next;
        }

        public async Task Invoke(HttpContext httpContext)
        {
            var response = httpContext.Items.DownstreamResponse();
            const string streamingContent = "text/event-stream";
            if (httpContext.Response.Headers["Content-Type"] == streamingContent
                || httpContext.Response.ContentType == streamingContent)
            {
                //await ForwardStream(httpContext.Response.Body, response);
                var stream = await response.Content.ReadAsStreamAsync();
                await WriteServerStreamEventsAsync(httpContext, stream);
            }
            await _next.Invoke(httpContext);
        }
        private static async Task WriteServerStreamEventsAsync(HttpContext httpContext, Stream contentStream)
        {
            var responseStream = httpContext.Response.Body;
            await contentStream.CopyToAsync(responseStream, httpContext.RequestAborted);
            await responseStream.FlushAsync(httpContext.RequestAborted);
        }
 }

[raman-m]

Dear @alivy !
The official language of the project is English. Most of our teammates and even our contributors/developers from all over the world do not understand Chinese!

[ggnaegi]

When there is text/event-stream in the response body, the returned result is loaded and returned at once, not refreshed in real time.
Even when I load my own Middleware and add it to OcelotPipelineExtensions, it doesn't refresh in real time, and the results are received as an asynchronous stream.
The custom extension code is as follows

[ggnaegi]

@alivy @raman-m 我不懂中文，所以可能无法用中文表达清楚。I would like to learn your language though. I think, you should wait the latest changes that were pushed to the develop branch. We are not copying the request body and response body directly anymore.

[alivy]

@alivy @raman-m 我不懂中文，所以可能无法用中文表达清楚。不过我想学习你们的语言。我认为，您应该等待推送到开发分支的最新更改。我们不再直接复制请求正文和响应正文。

Due to environmental factors, the code cannot be made public, but I can describe the scenario to you ：
When I used Ocelot to handle OpenAI's dialogue interface v1/chat/completion, the direct request was to refresh each field in real-time and return it in asynchronous flow form. However, when I used the Ocelot gateway, the response result was also returned in asynchronous flow form, but all dialogue information was returned at once

[raman-m]

Due to environmental factors, the code cannot be made public

Are you working for the government? It is essential to understand that you are writing in an open-source code product, and we can easily impose a ban on you. It is necessary for you to demonstrate your willingness to share the code!

[alivy]

请使用英文文本更新描述，否则问题将被关闭！

The scene is as follows:
When I used Ocelot to handle OpenAI's dialogue interface v1/chat/completion, the direct request was to refresh each field in real-time and return it in asynchronous flow form. However, when I used the Ocelot gateway, the response result was also returned in asynchronous flow form, but all dialogue information was returned at once

[alivy]

When I tried to modify the addition of middleware in the pipeline to respond to data asynchronously, an error occurred;

 public class CustomResponseMiddleware : OcelotMiddleware
 {

     private readonly RequestDelegate _next;
     public CustomResponseMiddleware(RequestDelegate next, IOcelotLoggerFactory loggerFactory)
         : base(loggerFactory.CreateLogger<CustomResponseMiddleware>())
     {
         _next = next;
     }

     public async Task Invoke(HttpContext httpContext)
     {
         var response = httpContext.Items.DownstreamResponse();
         const string streamingContent = "text/event-stream";
         if (httpContext.Response.Headers["Content-Type"] == streamingContent
             || httpContext.Response.ContentType == streamingContent
             || response.Content.Headers.ContentType?.ToString() == streamingContent)
         {
             //await ForwardStream(httpContext.Response.Body, response);
             var stream = await response.Content.ReadAsStreamAsync();
             await WriteServerStreamEventsAsync(httpContext, stream);
         }
         await _next.Invoke(httpContext);
     }
     private static async Task WriteServerStreamEventsAsync(HttpContext httpContext, Stream contentStream)
     {
         var responseStream = httpContext.Response.Body;
         await contentStream.CopyToAsync(responseStream, httpContext.RequestAborted);
         await responseStream.FlushAsync(httpContext.RequestAborted);
     }
 }

The error message is as follows：

Microsoft.AspNetCore.Server.Kestrel[13]
Connection id "0HN0RN6PE4IOV", Request id "0HN0RN6PE4IOV:00000002": An unhandled exception was thrown by the application.
System.InvalidOperationException: Response Content-Length mismatch: too few bytes written (0 of 71354).

[raman-m]

Dear @alivy !
Seems you try to use Server-Sent Events protocol in Ocelot, because I see text/event-stream media type (content type) in your code snippet.
⚠️ Please note! Ocelot doesn't support SSE protocol officially!
Your researches should be done by your own.
Ocelot has 2 different system pipelines: http and ws ones, aka HTTP(s) pipeline and Websocket pipeline.
You try to attach custom middleware which is not suitable either to Http or to WebSocket pipeline because of system ASP.NET services mapper settings.
To start supporting SSE protocol we have to develop 3rd pipeline for text/event-stream content type streaming!!!

Finally, I would say your experiments with custom middleware to fix or tune behavior of Http pipeline will probably fail. We don't guarantee a success of such researches because of written reasons above!

@ggnaegi FYI

[alivy]

Thank you for your reply. I hope to support the sse protocol in the future to make using Ocelot more comprehensive

[raman-m]

Your custom delegate analyzes the content of bodies. Why do you develop such solutions?
It is inappropriate to observe people through door keyhole! 😄

[ggnaegi]

@alivy could you tell us which Ocelot version you are using?

[raman-m]

Seems, the latest one...

[alivy]

@alivy您能告诉我们您使用的是哪个 Ocelot 版本吗？

Ocelot 22.0.1
My project uses .net6

[raman-m]

@alivy @ggnaegi
Please, continue your nice discussion in this thread!

[alivy]

You can test it using an asynchronous flow interface, similar to the OpenAI session interface of ChatGpt. It seems that it is not possible to asynchronously refresh the returned data word by word, but rather wait for the asynchronous flow to load before returning the data. I guess there may be a problem with the Middleware processing of the OcelotPipelineExtensions

[raman-m]

I don't care about your love of any AI including ChatGPT !
I have no time for any tests for extra researches. Please, do your any researches by your own.
Please read my official answer to you here above!

Please continue discussion with @ggnaegi! You can write in Chinese talking to Guillaume... 😉

[alivy]

Don't misunderstand, I just used ChatGpt to express the scenarios of the problems I encountered, maybe my English expression is not very good

[ggnaegi]

@alivy @raman-m it seems we need to disable buffering on the response:
HttpContext.Features.Get<IHttpResponseBodyFeature>().DisableBuffering()
https://learn.microsoft.com/en-us/dotnet/api/microsoft.aspnetcore.http.features.ihttpresponsebodyfeature?view=aspnetcore-8.0

[alivy]

I have read the Ocelot source code and may need to make a judgment. If the response type is "text/event stream", it needs to be in the "HttpContextResponse. SetResponseOnHttpContext" method
Add the code "await htpContext. Response. Body. FlushAsync (context. RequestAborted);" to achieve asynchronous stream refresh;

https://learn.microsoft.com/en-us/dotnet/api/microsoft.aspnetcore.webutilities.httpresponsestreamwriter.flushasync?view=aspnetcore-8.0

[ggnaegi]

@alivy yes, with the response buffering disabled, then it might work. So you would need to update the ocelot builder and SetResponseOnHttpContext. What about creating your own PR and make it work? thanks

[raman-m]

I'm sure on 100% we have to have 3rd pipeline for the protocol. Current pipelines will not be changed at all!
So, why not to create a PR and discuss real but draft solution instead of long discussions here...
But sure this tuned draft solution should be compiled and Ocelot should start without errors 😉

[raman-m]

The typical case... When you asked the people to contribute they disappear... :))

@alivy I like pandas! What do you know about pandas? 😉

@ggnaegi I think we must have different ASP.NET pipeline of composed middlewares. And we have to detect SSE request on early stage blocking to start our main Ocelot pipeline. Also, it will be bad to extend current pipeline making it complex and resource consuming.
But in theory we can reuse some middlewares for sure.
Let's return to SSE tasks after a couple of releases...

[CodeABao]

@alivy do you have a solution for this issue?

[CodeABao]

@raman-m sorry，I am not familiar with SSE, i will modify souce code to do some tests.

[ggnaegi]

@alivy it would help us if you could share some parts of your code with us

[raman-m]

@ggnaegi Silence means it's raining in China ))

[alivy]

Sorry, due to the sluggish market, I have resigned from my previous company and am currently unable to provide the code. I will organize and submit the code when I have time

[raman-m]

Yea, in the next life probably...
Congrats in changing your lord! 😄 Are you free man now? 😃

[EnginKorkmaz]

app.UseOcelot().Wait();

const string STREAMING_CONTENT = "text/event-stream";

app.Use(async (context, next) => {
    DownstreamResponse response = context.Items.DownstreamResponse();
    if (context.Response.Headers["Content-Type"] == STREAMING_CONTENT
        || context.Response.ContentType == STREAMING_CONTENT
        || response?.Content.Headers.ContentType?.ToString() == STREAMING_CONTENT) {
        context.Features.Get<IHttpResponseBodyFeature>()?.DisableBuffering();
    }
    await next();
});

app.Run();

The code above works for me.

[raman-m]

@EnginKorkmaz Hello Engin!
This coding recipe quite straightforward and can be easily converted into middleware. Does this solution support bi-directional mode? How thoroughly it tested and utilized in a environment? If you are confident in its effectiveness, why not consider opening a pull request?

[raman-m]

@ggnaegi wrote on Jan 23

HttpContext.Features.Get<IHttpResponseBodyFeature>().DisableBuffering()

There are similar proposals mentioned above. I believe it is time to open a low-priority PR and prioritize it when backlog has been cleared.

[EnginKorkmaz]

Hello @raman-m ,
This straightforward solution seems to address my current scenario effectively. I haven't had the opportunity to test its bidirectional support yet, as I have only tried it in our application that uses Server-Sent Events. Large-scale testing hasn't been conducted either, so we plan to monitor its performance after deploying it into production. If no issues arise, we can consider opening a PR. Additionally, I have updated Ocelot to the latest version.