sse_handler.tpl 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768
  1. // Code scaffolded by goctl. Safe to edit.
  2. // goctl {{.version}}
  3. package {{.PkgName}}
  4. import (
  5. "encoding/json"
  6. "fmt"
  7. "net/http"
  8. "github.com/zeromicro/go-zero/core/logc"
  9. "github.com/zeromicro/go-zero/core/threading"
  10. {{if .HasRequest}}"github.com/zeromicro/go-zero/rest/httpx"{{end}}
  11. {{.ImportPackages}}
  12. )
  13. {{if .HasDoc}}{{.Doc}}{{end}}
  14. func {{.HandlerName}}(svcCtx *svc.ServiceContext) http.HandlerFunc {
  15. return func(w http.ResponseWriter, r *http.Request) {
  16. {{if .HasRequest}}var req types.{{.RequestType}}
  17. if err := httpx.Parse(r, &req); err != nil {
  18. httpx.ErrorCtx(r.Context(), w, err)
  19. return
  20. }
  21. {{end}}// Buffer size of 16 is chosen as a reasonable default to balance throughput and memory usage.
  22. // You can change this based on your application's needs.
  23. // if your go-zero version less than 1.8.1, you need to add 3 lines below.
  24. // w.Header().Set("Content-Type", "text/event-stream")
  25. // w.Header().Set("Cache-Control", "no-cache")
  26. // w.Header().Set("Connection", "keep-alive")
  27. client := make(chan {{.ResponseType}}, 16)
  28. l := {{.LogicName}}.New{{.LogicType}}(r.Context(), svcCtx)
  29. threading.GoSafeCtx(r.Context(), func() {
  30. defer close(client)
  31. err := l.{{.Call}}({{if .HasRequest}}&req, {{end}}client)
  32. if err != nil {
  33. logc.Errorw(r.Context(), "{{.HandlerName}}", logc.Field("error", err))
  34. return
  35. }
  36. })
  37. for {
  38. select {
  39. case data, ok := <-client:
  40. if !ok {
  41. return
  42. }
  43. output, err := json.Marshal(data)
  44. if err != nil {
  45. logc.Errorw(r.Context(), "{{.HandlerName}}", logc.Field("error", err))
  46. continue
  47. }
  48. if _, err := fmt.Fprintf(w, "data: %s\n\n", string(output)); err != nil {
  49. logc.Errorw(r.Context(), "{{.HandlerName}}", logc.Field("error", err))
  50. return
  51. }
  52. if flusher, ok := w.(http.Flusher); ok {
  53. flusher.Flush()
  54. }
  55. case <-r.Context().Done():
  56. return
  57. }
  58. }
  59. }
  60. }