From c5ae25408db4ae691dd24f765e1a6ed3dcb29ccd Mon Sep 17 00:00:00 2001 From: junevanlong Date: Mon, 22 Jun 2026 19:13:01 +0800 Subject: [PATCH] examples: add prewarm and SSE examples --- docs/user_guide/server/overview.md | 27 +++++++++ docs/user_guide/server/overview.zh_CN.md | 20 +++++++ examples/features/prewarm/README.md | 36 +++++++++++ examples/features/prewarm/client/main.go | 51 ++++++++++++++++ examples/features/prewarm/server/main.go | 38 ++++++++++++ examples/features/prewarm/server/trpc_go.yaml | 8 +++ examples/features/sse/README.md | 32 ++++++++++ examples/features/sse/client/main.go | 53 ++++++++++++++++ examples/features/sse/server/main.go | 60 +++++++++++++++++++ examples/go.mod | 2 +- 10 files changed, 326 insertions(+), 1 deletion(-) create mode 100644 examples/features/prewarm/README.md create mode 100644 examples/features/prewarm/client/main.go create mode 100644 examples/features/prewarm/server/main.go create mode 100644 examples/features/prewarm/server/trpc_go.yaml create mode 100644 examples/features/sse/README.md create mode 100644 examples/features/sse/client/main.go create mode 100644 examples/features/sse/server/main.go diff --git a/docs/user_guide/server/overview.md b/docs/user_guide/server/overview.md index eac2b36e..940a9caa 100644 --- a/docs/user_guide/server/overview.md +++ b/docs/user_guide/server/overview.md @@ -262,6 +262,33 @@ tRPC-Go has considered the testability of the framework from the beginning of th # Advanced Features +## Precool Readiness Checks + +Precool checks are used when a process has started but some dependencies still +need to finish initialization before the service should be considered ready. +The admin endpoint `/cmds/is_precool/` reports process-level status, and +`/cmds/is_precool/{service-name}` reports service-level status. + +When the server is created with `trpc.NewServer()`, tRPC-Go creates a precool +checker and wires it to both the server and admin service. Register service +readiness logic with `RegisterServicePrecool`: + +```go +s := trpc.NewServer() +if err := s.RegisterServicePrecool("trpc.examples.precool.Precool", func() precool.Status { + if !dependencyReady { + return precool.Ongoing + } + return precool.Success +}); err != nil { + return err +} +``` + +If a server or admin service is constructed manually, pass the same +`precool.Checker` instance with `server.WithPrecool` and `admin.WithPrecool`. +See `examples/features/precool` for a complete runnable example. + ## Timeout control tRPC-Go provides three timeout mechanisms for RPC calls: link timeout, message timeout, and call timeout. For an introduction to the principles and related configurations of these three timeout mechanisms, please refer to [tRPC-Go Timeout Control](/docs/user_guide/timeout_control.md). diff --git a/docs/user_guide/server/overview.zh_CN.md b/docs/user_guide/server/overview.zh_CN.md index 7c74efdf..fc4221a9 100644 --- a/docs/user_guide/server/overview.zh_CN.md +++ b/docs/user_guide/server/overview.zh_CN.md @@ -262,6 +262,26 @@ tRPC-Go 从设计之初就考虑了框架的易测性，在通过 pb 生成桩 # 高级功能 +## 预冷就绪检查 + +预冷检查适用于进程已经启动，但部分依赖还需要完成初始化，服务暂时不应被认为完全就绪的场景。Admin 接口 `/cmds/is_precool/` 返回进程级状态，`/cmds/is_precool/{service-name}` 返回服务级状态。 + +使用 `trpc.NewServer()` 创建服务时，tRPC-Go 会创建一个 precool checker，并同时接入 server 和 admin。业务可以通过 `RegisterServicePrecool` 注册服务级就绪逻辑： + +```go +s := trpc.NewServer() +if err := s.RegisterServicePrecool("trpc.examples.precool.Precool", func() precool.Status { + if !dependencyReady { + return precool.Ongoing + } + return precool.Success +}); err != nil { + return err +} +``` + +如果手动构造 server 或 admin，则需要用 `server.WithPrecool` 和 `admin.WithPrecool` 传入同一个 `precool.Checker` 实例。完整可运行示例见 `examples/features/precool`。 + ## 超时控制 tRPC-Go 为 RPC 调用提供了 3 种超时机制控制：链路超时，消息超时和调用超时。关于这 3 种超时机制的原理介绍和相关配置，请参考 [tRPC-Go 超时控制](/docs/user_guide/timeout_control.zh_CN.md)。 diff --git a/examples/features/prewarm/README.md b/examples/features/prewarm/README.md new file mode 100644 index 00000000..f330c2e4 --- /dev/null +++ b/examples/features/prewarm/README.md @@ -0,0 +1,36 @@ +# Prewarm + +This example demonstrates how to explicitly initialize a client and prewarm its +connection pool before sending requests. + +## Usage + +* Start server. + +```shell +$ go run server/main.go -conf server/trpc_go.yaml +``` + +* Start client. + +```shell +$ go run client/main.go +``` + +The client initializes `client.DefaultClient` with `client.WithPreWarm` before +creating the generated proxy. If initialization succeeds, the connection pool +has already established the configured number of connections for the target +node, and the following RPC can reuse the warmed pool. + +## Explanation + +Prewarming is opt-in. Set `transport.PreWarmOptions.ConnsPerNode` to a positive +value to enable it. `Timeout` bounds the initialization phase. + +This example only uses public APIs: + +* `client.InitializableClient` +* `client.WithPreWarm` +* `transport.PreWarmOptions` + +It does not inspect internal pool metrics or connection counters. diff --git a/examples/features/prewarm/client/main.go b/examples/features/prewarm/client/main.go new file mode 100644 index 00000000..f3b33ad4 --- /dev/null +++ b/examples/features/prewarm/client/main.go @@ -0,0 +1,51 @@ +// +// +// Tencent is pleased to support the open source community by making tRPC available. +// +// Copyright (C) 2023 Tencent. +// All rights reserved. +// +// If you have downloaded a copy of the tRPC source code from Tencent, +// please note that tRPC source code is licensed under the Apache 2.0 License, +// A copy of the Apache 2.0 License is included in this file. +// +// + +// Package main is the client main package for prewarm demo. +package main + +import ( + "context" + "time" + + "trpc.group/trpc-go/trpc-go/client" + "trpc.group/trpc-go/trpc-go/examples/helloworld/pb" + "trpc.group/trpc-go/trpc-go/log" + "trpc.group/trpc-go/trpc-go/transport" +) + +func main() { + ctx := context.Background() + opts := []client.Option{ + client.WithTarget("ip://127.0.0.1:8000"), + client.WithPreWarm(transport.PreWarmOptions{ + ConnsPerNode: 2, + Timeout: time.Second, + }), + } + + initializable, ok := client.DefaultClient.(client.InitializableClient) + if !ok { + log.Fatal("default client does not support initialization") + } + if err := initializable.Init(ctx, opts...); err != nil { + log.Fatalf("prewarm client: %v", err) + } + + proxy := pb.NewGreeterClientProxy(opts...) + rsp, err := proxy.Hello(ctx, &pb.HelloRequest{Msg: "prewarm"}) + if err != nil { + log.Fatalf("hello: %v", err) + } + log.Infof("hello response: %s", rsp.Msg) +} diff --git a/examples/features/prewarm/server/main.go b/examples/features/prewarm/server/main.go new file mode 100644 index 00000000..73ffeaf6 --- /dev/null +++ b/examples/features/prewarm/server/main.go @@ -0,0 +1,38 @@ +// +// +// Tencent is pleased to support the open source community by making tRPC available. +// +// Copyright (C) 2023 Tencent. +// All rights reserved. +// +// If you have downloaded a copy of the tRPC source code from Tencent, +// please note that tRPC source code is licensed under the Apache 2.0 License, +// A copy of the Apache 2.0 License is included in this file. +// +// + +// Package main is the server main package for prewarm demo. +package main + +import ( + "context" + + trpc "trpc.group/trpc-go/trpc-go" + "trpc.group/trpc-go/trpc-go/examples/helloworld/pb" + "trpc.group/trpc-go/trpc-go/log" +) + +func main() { + s := trpc.NewServer() + pb.RegisterGreeterService(s, greeter{}) + if err := s.Serve(); err != nil { + log.Error(err) + } +} + +type greeter struct{} + +func (greeter) Hello(ctx context.Context, req *pb.HelloRequest) (*pb.HelloReply, error) { + log.Infof("got hello request: %s", req.Msg) + return &pb.HelloReply{Msg: "Hello " + req.Msg + "!"}, nil +} diff --git a/examples/features/prewarm/server/trpc_go.yaml b/examples/features/prewarm/server/trpc_go.yaml new file mode 100644 index 00000000..54ec9841 --- /dev/null +++ b/examples/features/prewarm/server/trpc_go.yaml @@ -0,0 +1,8 @@ +server: + service: + - name: trpc.helloworld.Greeter + ip: 127.0.0.1 + port: 8000 + network: tcp + protocol: trpc + timeout: 1000 diff --git a/examples/features/sse/README.md b/examples/features/sse/README.md new file mode 100644 index 00000000..cb6fcdb9 --- /dev/null +++ b/examples/features/sse/README.md @@ -0,0 +1,32 @@ +# Server-Sent Events + +This example demonstrates how to consume server-sent events with the tRPC-Go +HTTP client codec and how to write SSE events with `http.WriteSSE`. + +## Usage + +* Start server. + +```shell +$ go run server/main.go +``` + +* Start client. + +```shell +$ go run client/main.go +``` + +The client sets `http.ClientRspHeader.SSEHandler`. When the response is an SSE +stream, the framework parses each event and invokes the handler. + +## Explanation + +The server is a small standard-library HTTP server so the example stays focused +on the public SSE APIs: + +* `http.WriteSSE` +* `http.ClientRspHeader.SSEHandler` +* `http.ClientRspHeader.SSECondition` + +No additional transport plugin or framework extension is required. diff --git a/examples/features/sse/client/main.go b/examples/features/sse/client/main.go new file mode 100644 index 00000000..c125a9c5 --- /dev/null +++ b/examples/features/sse/client/main.go @@ -0,0 +1,53 @@ +// +// +// Tencent is pleased to support the open source community by making tRPC available. +// +// Copyright (C) 2023 Tencent. +// All rights reserved. +// +// If you have downloaded a copy of the tRPC source code from Tencent, +// please note that tRPC source code is licensed under the Apache 2.0 License, +// A copy of the Apache 2.0 License is included in this file. +// +// + +// Package main is the client main package for SSE demo. +package main + +import ( + "context" + stdhttp "net/http" + "strings" + + "github.com/r3labs/sse/v2" + + "trpc.group/trpc-go/trpc-go/client" + "trpc.group/trpc-go/trpc-go/codec" + thttp "trpc.group/trpc-go/trpc-go/http" + "trpc.group/trpc-go/trpc-go/log" +) + +type eventHandler struct{} + +func (eventHandler) Handle(event *sse.Event) error { + log.Infof("event id=%s type=%s data=%s", event.ID, event.Event, event.Data) + return nil +} + +func main() { + proxy := thttp.NewClientProxy( + "trpc.examples.sse.Events", + client.WithTarget("ip://127.0.0.1:8080"), + client.WithCurrentSerializationType(codec.SerializationTypeNoop), + ) + + rspHead := &thttp.ClientRspHeader{ + SSECondition: func(rsp *stdhttp.Response) bool { + return strings.Contains(rsp.Header.Get("Content-Type"), "text/event-stream") + }, + SSEHandler: eventHandler{}, + } + if err := proxy.Get(context.Background(), "/events", nil, client.WithRspHead(rspHead)); err != nil { + log.Fatalf("get events: %v", err) + } +} diff --git a/examples/features/sse/server/main.go b/examples/features/sse/server/main.go new file mode 100644 index 00000000..1144108e --- /dev/null +++ b/examples/features/sse/server/main.go @@ -0,0 +1,60 @@ +// +// +// Tencent is pleased to support the open source community by making tRPC available. +// +// Copyright (C) 2023 Tencent. +// All rights reserved. +// +// If you have downloaded a copy of the tRPC source code from Tencent, +// please note that tRPC source code is licensed under the Apache 2.0 License, +// A copy of the Apache 2.0 License is included in this file. +// +// + +// Package main is the server main package for SSE demo. +package main + +import ( + "fmt" + stdhttp "net/http" + "strconv" + "time" + + "github.com/r3labs/sse/v2" + + thttp "trpc.group/trpc-go/trpc-go/http" +) + +func main() { + mux := stdhttp.NewServeMux() + mux.HandleFunc("/events", events) + + fmt.Println("SSE server listening on http://127.0.0.1:8080/events") + if err := stdhttp.ListenAndServe("127.0.0.1:8080", mux); err != nil { + fmt.Println(err) + } +} + +func events(w stdhttp.ResponseWriter, _ *stdhttp.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + + flusher, ok := w.(stdhttp.Flusher) + if !ok { + stdhttp.Error(w, "streaming unsupported", stdhttp.StatusInternalServerError) + return + } + + for i := 1; i <= 3; i++ { + event := sse.Event{ + ID: []byte(strconv.Itoa(i)), + Event: []byte("message"), + Data: []byte(fmt.Sprintf("event-%d", i)), + } + if err := thttp.WriteSSE(w, event); err != nil { + return + } + flusher.Flush() + time.Sleep(200 * time.Millisecond) + } +} diff --git a/examples/go.mod b/examples/go.mod index 05b6a467..e28f9223 100644 --- a/examples/go.mod +++ b/examples/go.mod @@ -6,6 +6,7 @@ replace trpc.group/trpc-go/trpc-go => ../ require ( github.com/golang/protobuf v1.5.2 + github.com/r3labs/sse/v2 v2.10.0 github.com/valyala/fasthttp v1.43.0 google.golang.org/protobuf v1.33.0 trpc.group/trpc-go/trpc-go v0.0.0-00010101000000-000000000000 @@ -32,7 +33,6 @@ require ( github.com/panjf2000/ants/v2 v2.4.6 // indirect github.com/pierrec/lz4/v4 v4.1.21 // indirect github.com/pkg/errors v0.9.1 // indirect - github.com/r3labs/sse/v2 v2.10.0 // indirect github.com/spf13/cast v1.3.1 // indirect github.com/valyala/bytebufferpool v1.0.0 // indirect go.uber.org/atomic v1.11.0 // indirect