text/event-stream not supported

[rome-user]

What version of ogen are you using?

$ go list -m github.com/ogen-go/ogen
github.com/ogen-go/ogen v1.8.1

Can this issue be reproduced with the latest version?

Yes

What did you do?

Consider the following schema. Save it as openapi.yaml.

openapi: 3.0.0
info:
  title: Example API
  version: "1.0"
paths:
  /events:
    get:
      summary: SSE example
      responses:
        "200":
          description: OK
          content:
            text/event-stream:
              schema:
                type: string

Run the following terminal commands

$ go mod init repro
$ go get github.com/ogen-go/ogen/cmd/ogen
$ go run github.com/ogen-go/ogen/cmd/ogen --target gen/api -package api --clean openapi.yaml
$ go mod tidy

Create a main.go file with the following contents.

Click to view server implementation

package main

import (
	"context"
	"io"
	"log"
	"net/http"
	"time"

	"repro/api"
)

func main() {
	srv, err := api.NewServer(&service{})
	if err != nil {
		log.Fatal(err)
	}

	if err := http.ListenAndServe(":8080", srv); err != nil {
		log.Fatal(err)
	}
}

type service struct{}

func (s *service) EventsGet(ctx context.Context) (api.EventsGetOK, error) {
	r, w := io.Pipe()

	go func() {
		defer w.Close()
		for ctx.Err() == nil {
			select {
			case <-ctx.Done():
				return
			case <-time.After(1 * time.Second):
				w.Write([]byte("data: ping\n\n"))
			}
		}
	}()

	return api.EventsGetOK{Data: r}, nil
}

Finally, start the server.

$ go run main.go

What did you expect to see?

Client can stream the response. e.g.

$ curl -N http://localhost:8080/events
data: ping
data: ping
...

What did you see instead?

The generated code in oas_response_encoders_gen.go looks like the following.

Click to view generated code

func encodeEventsGetResponse(response EventsGetOK, w http.ResponseWriter, span trace.Span) error {
	w.Header().Set("Content-Type", "text/event-stream")
	w.WriteHeader(200)
	span.SetStatus(codes.Ok, http.StatusText(200))

	writer := w
	if _, err := io.Copy(writer, response); err != nil {
		return errors.Wrap(err, "write")
	}

	return nil
}

The call to io.Copy prevents streaming of response. This results in clients not receiving data.

[rokf]

I'm personally interested in this functionality so that I could generate boilerplate handler code for https://data-star.dev with ogen.

What I do currently is that I have two OpenAPI specifications for an application - one API that returns HTML for the GUI and one that returns JSON for M2M communication.

With support for text/event-stream I could squeeze in the SSE endpoints for Datastar into my ogen HTML API - currently I need a custom set of handlers next to those managed by ogen and combine them all with a mux.

[rokf]

As a workaround you can use middleware to pass and retrieve the raw request and response objects - #1252.

I've tested this approach with https://github.com/starfederation/datastar/tree/main/sdk/go and it works fine.

The only issue I found was that Ogen writes an error log saying that there's an superfluous header write once the stream is finished - it doesn't seem to break anything though.

[rokf]

I made another discovery today, which makes the current workaround totally usable as it seems - the superfluous header error is gone 😄

You'll need at least one dummy response in the OpenAPI specification:

responses:
        "default":
          $ref: "#/components/responses/default"

Then you can use a dummy no-op error and watch for it in Ogen's error handler, i.e.:

var Noop = errors.New("noop-error")

func ErrorHandler(ctx context.Context, w http.ResponseWriter, r *http.Request, err error) {
	if errors.Is(err, Noop) {
		return
	}

In the Ogen HTTP handlers where manually writing your responses you can then return the no-op error at the end of the SSE "session" like:

        r := util.RequestFromContext(ctx)
	w := util.ResponseFromContext(ctx)

        // ...

	sse := datastar.NewSSE(w, r)
	err = sse.MergeFragments(buff.String())
	if err != nil {
		return nil, err
	}

	return nil, herrors.Noop
}

[joeblew999]

I made another discovery today, which makes the current workaround totally usable as it seems - the superfluous header error is gone 😄

You'll need at least one dummy response in the OpenAPI specification:

responses:
        "default":
          $ref: "#/components/responses/default"

Then you can use a dummy no-op error and watch for it in Ogen's error handler, i.e.:

var Noop = errors.New("noop-error")

func ErrorHandler(ctx context.Context, w http.ResponseWriter, r *http.Request, err error) {
if errors.Is(err, Noop) {
return
}
In the Ogen HTTP handlers where manually writing your responses you can then return the no-op error at the end of the SSE "session" like:

    r := util.RequestFromContext(ctx)

w := util.ResponseFromContext(ctx)

    // ...

sse := datastar.NewSSE(w, r)
err = sse.MergeFragments(buff.String())
if err != nil {
return nil, err
}

return nil, herrors.Noop
}

Hey @rokf

I also use Datastar and also have the exact same need in that I need a HTML API and a M2M API.

If you have a repo for this, I would love to help out.

[rokf]

@joeblew999 the repo in which we're using patterns described above is private. It's for a project at work. I don't have a public repo with this code.

Regarding Datastar and its use together with ogen - you should be able to find all the information you'll need to make it work in this thread as long as you're familiar with the basics of ogen.

For making a HTTP server that combines two or more HTTP servers generated by ogen you can use http.NewServeMux and its Handle method to attach server instances under different prefixes. You might have to strip prefixes with http.StripPrefix - it depends on your endpoints in the OpenAPI specifications.