inanzzz | Golang'da gerçek zamanlı iletişim için Server Sent Event örneği

Sunucudan Gönderilen Olay (SSE) teknolojisi, verileri (metin/dize) web sunucusundan tarayıcıya "gerçek zamanlı" olarak göndermek için tek bir uzun ömürlü HTTP/1.1 veya HTTP/2 bağlantısı kullanır. WebSockets teknolojisine benzer ancak tek yönlü bir modeldir. İstemci (tarayıcı), EventSource arayüzünü kullanarak bir sunucuya bağlantı oluşturur ve sunucu bu bağlantıyı kullanarak verileri geri aktarır. Bağlantı herhangi bir nedenle kesilirse EventSource otomatik olarak yeniden bağlanmayı dener. Eşzamanlı bağlantı limitleriyle (tarayıcı başına açık sekmeler) ilgili özel bir sınırlama vardır. Yazma sırasında, HTTP/1.1 için ortak sınır 6 ve HTTP/2 için yaklaşık 100'dür (yapılandırılabilir).

Bu örnekte, veritabanını doldurmak için bir yazma uç noktasına istekler göndereceğiz. Sahne arkasında, okuma uç noktası bu verileri listeleme amacıyla tarayıcıya geri aktarır.

Dosyalar

index.html

<!DOCTYPE html>
<html>
    <head>
        <title>Audit events</title>
        <script>
            const endpoint = new EventSource("http://localhost:8888/api/v1/audits");

            endpoint.onmessage = function(audit) {
                const container = document.getElementById("records");
                const record = document.createElement("p");

                container.append(audit.data, record);
            };
        </script>
    </head>
    <body>
        <div id="records"></div>
    </body>
</html>

main.go

package main

import (
	"net/http"
	"time"

	"sse/api"
	"sse/storage"
)

func main() {
	auditStorage := storage.NewAudit()
	auditHandler := api.NewAudit(auditStorage, time.Second)

	rtr := http.NewServeMux()
	rtr.HandleFunc("POST /api/v1/audits", auditHandler.Write)
	rtr.HandleFunc("GET /api/v1/audits", auditHandler.Read)

	http.ListenAndServe(":8888", rtr)
}

api/audit.go

package api

import (
	"encoding/json"
	"fmt"
	"net/http"
	"time"

	"sse/model"
)

type storer interface {
	Insert(audit *model.Audit)
	List(since time.Time) []*model.Audit
}

type Audit struct {
	storage storer
	delay   time.Duration
}

func NewAudit(str storer, delay time.Duration) Audit {
	return Audit{
		storage: str,
		delay:   delay,
	}
}

func (a Audit) Write(w http.ResponseWriter, r *http.Request) {
	var req model.Request

	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		fmt.Println(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	a.storage.Insert(&model.Audit{
		ID:        req.ID,
		CreatedAt: time.Now(),
	})
}

func (a Audit) Read(w http.ResponseWriter, r *http.Request) {
	flusher, ok := w.(http.Flusher)
	if !ok {
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	w.Header().Set("Access-Control-Allow-Origin", "*") // In prod, only allow known origins!
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("Content-Type", "text/event-stream")

	since := time.Now()

	for {
		audits := a.storage.List(since)
		since = time.Now()

		for _, audit := range audits {
			res, err := json.Marshal(model.Response{
				ID:        audit.ID,
				CreatedAt: audit.CreatedAt,
			})
			if err != nil {
				fmt.Println(err)
				continue
			}

			fmt.Fprintf(w, "data: %v\n\n", string(res))
			flusher.Flush()
		}

		<-time.After(a.delay)
	}

	// TODO: Make sure to disconnect as per your feature needs. I am skipping it.
}

model/api.go

package model

import "time"

type Request struct {
	ID string `json:"id"`
}

type Response struct {
	ID        string    `json:"id"`
	CreatedAt time.Time `json:"created_at"`
}

model/storage.go

package model

import "time"

type Audit struct {
	ID        string
	CreatedAt time.Time
}

storage/audit.go

package storage

import (
	"sync"
	"time"

	"sse/model"
)

type Audit struct {
	records *sync.Map
}

func NewAudit() Audit {
	return Audit{
		records: &sync.Map{},
	}
}

func (a Audit) Insert(audit *model.Audit) {
	a.records.Store(audit.ID, audit)
}

func (a Audit) List(since time.Time) []*model.Audit {
	var list []*model.Audit

	a.records.Range(func(k, v any) bool {
		audit := v.(*model.Audit)
		if audit.CreatedAt.Equal(since) || audit.CreatedAt.After(since) {
			list = append(list, audit)
		}

		return true
	})

	return list
}

Test

index.html'i tarayıcı(lar)da dilediğiniz kadar sekmeyle açın. Uygulamayı go run -race main.go komutuyla çalıştırın ve istekleri write uç noktasına göndermek için aşağıdaki komutu kullanın.

$ for i in `seq 1 10`; do curl -i -X POST -d '{"id": "'$i'"}' http://localhost:8888/api/v1/audits ; done

Tarayıcıya baktığınızda aşağıdaki gibi bir çıktı görmelisiniz.

{"id":"1","created_at":"2024-06-09T16:01:57.770701+01:00"}
{"id":"2","created_at":"2024-06-09T16:01:57.801793+01:00"}
{"id":"3","created_at":"2024-06-09T16:01:58.803549+01:00"}
{"id":"4","created_at":"2024-06-09T16:01:59.800784+01:00"}
{"id":"5","created_at":"2024-06-09T16:02:00.798023+01:00"}
{"id":"6","created_at":"2024-06-09T16:02:01.797871+01:00"}
{"id":"7","created_at":"2024-06-09T16:02:02.799884+01:00"}
{"id":"8","created_at":"2024-06-09T16:02:03.801665+01:00"}
{"id":"9","created_at":"2024-06-09T16:02:04.809286+01:00"}
{"id":"10","created_at":"2024-06-09T16:02:05.803576+01:00"}