Merge pull request #8 from go-zoox/feat/add-pubsub

feat: add pubsub

application.go

@@ -25,6 +25,8 @@ import (
  "github.com/go-zoox/zoox/components/application/jobqueue"
  "github.com/go-zoox/zoox/components/application/runtime"
  "github.com/go-zoox/zoox/components/application/websocket"
+
+ "github.com/go-zoox/pubsub"
 )
 
 // HandlerFunc defines the request handler used by zoox
@@ -65,6 +67,11 @@ type Application struct {
  // Runtime
  runtime runtime.Runtime
 
+ //
+ jsonrpc jsonrpcServer.Server
+ //
+ pubsub pubsub.PubSub
+
  //
  Config ApplicationConfig
 }
@@ -95,6 +102,16 @@ type ApplicationConfig struct {
  Session session.Config `config:"session"`
  //
  Cache cache.Config `config:"cache"`
+ //
+ Redis Redis `config:"redis"`
+}
+
+type Redis struct {
+ Host string `config:"host"`
+ Port int `config:"port"`
+ DB int `config:"db"`
+ Username string `config:"username"`
+ Password string `config:"password"`
 }
 
 // New is the constructor of zoox.Application.
@@ -339,9 +356,13 @@ func (app *Application) IsProd() bool {
  return app.Env.Get("MODE") == "production"
 }
 
-// CreateJSONRPC creates a new CreateJSONRPC handler.
-func (app *Application) CreateJSONRPC(path string) jsonrpcServer.Server {
- rpc := jsonrpcServer.New()
+// JSONRPC get a new JSONRPC handler.
+func (app *Application) JSONRPC(path string) jsonrpcServer.Server {
+ if app.jsonrpc != nil {
+ return app.jsonrpc
+ }
+
+ app.jsonrpc = jsonrpcServer.New()
 
  app.Post(path, func(ctx *Context) {
  request, err := io.ReadAll(ctx.Request.Body)
@@ -351,7 +372,7 @@ func (app *Application) CreateJSONRPC(path string) jsonrpcServer.Server {
  }
  defer ctx.Request.Body.Close()
 
- response, err := rpc.Invoke(ctx.Context(), request)
+ response, err := app.jsonrpc.Invoke(ctx.Context(), request)
  if err != nil {
  ctx.Error(http.StatusInternalServerError, err.Error())
  return
@@ -361,7 +382,26 @@ func (app *Application) CreateJSONRPC(path string) jsonrpcServer.Server {
  ctx.Write(response)
  })
 
- return rpc
+ return app.jsonrpc
+}
+
+// PubSub get a new PubSub handler.
+func (app *Application) PubSub() pubsub.PubSub {
+ if app.Config.Redis.Host == "" {
+ panic("redis config is required for pubsub in application")
+ }
+
+ if app.pubsub == nil {
+ app.pubsub = pubsub.New(&pubsub.Config{
+ RedisHost: app.Config.Redis.Host,
+ RedisPort: app.Config.Redis.Port,
+ RedisDB: app.Config.Redis.DB,
+ RedisUsername: app.Config.Redis.Username,
+ RedisPassword: app.Config.Redis.Password,
+ })
+ }
+
+ return app.pubsub
 }
 
 // Cache ...

context.go

@@ -30,6 +30,7 @@ import (
  "github.com/go-zoox/zoox/components/context/state"
  "github.com/go-zoox/zoox/components/context/user"
  "github.com/go-zoox/zoox/utils"
+ "gocloud.dev/pubsub"
 
  "github.com/go-zoox/cookie"
  "github.com/go-zoox/core-utils/safe"
@@ -885,3 +886,13 @@ func (ctx *Context) BodyBytes() ([]byte, error) {
 
  return bytes, nil
 }
+
+// Publish publishes the message to the pubsub.
+func (ctx *Context) Publish(msg *pubsub.Message) error {
+ return ctx.App.PubSub().Publish(ctx.Context(), msg)
+}
+
+// Subscribe subscribes the topic with the handler.
+func (ctx *Context) Subscribe(topic string, handler pubsub.Handler) error {
+ return ctx.App.PubSub().Subscribe(ctx.Context(), topic, handler)
+}

go.mod

@@ -18,6 +18,7 @@ require (
  github.com/go-zoox/jwt v1.3.0
  github.com/go-zoox/logger v1.4.4
  github.com/go-zoox/proxy v1.4.1
+ github.com/go-zoox/pubsub v1.2.2
  github.com/go-zoox/random v1.0.4
  github.com/go-zoox/ratelimit v1.2.0
  github.com/go-zoox/session v1.2.0
@@ -45,6 +46,7 @@ require (
  github.com/google/uuid v1.3.0 // indirect
  github.com/pkg/errors v0.9.1 // indirect
  github.com/pmezard/go-difflib v1.0.0 // indirect
+ github.com/redis/go-redis/v9 v9.0.5 // indirect
  github.com/robfig/cron/v3 v3.0.1 // indirect
  github.com/spf13/cast v1.5.1 // indirect
  github.com/tidwall/gjson v1.15.0 // indirect

go.sum

@@ -1,3 +1,5 @@
+github.com/bsm/ginkgo/v2 v2.7.0 h1:ItPMPH90RbmZJt5GtkcNvIRuGEdwlBItdNVoyzaNQao=
+github.com/bsm/gomega v1.26.0 h1:LhQm+AFcgV2M0WyKroMASzAzCAJVpAxQXv4SaI9a69Y=
 github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
 github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
@@ -56,8 +58,8 @@ github.com/go-zoox/datetime v1.1.1/go.mod h1:os6lYW/GXNpCIseFrBr8DNcOiPuwl5Ttc/k
 github.com/go-zoox/debug v1.0.1 h1:lAsUnofJ1xWnfpHQQ6O5ss0xS0J+8LVAYbv+qiH0ARQ=
 github.com/go-zoox/debug v1.0.1/go.mod h1:7HvnBeV1dVuuGVnXSLdJ5OE6X/wIXAjIEyiaA7NqQPA=
 github.com/go-zoox/dotenv v1.0.7/go.mod h1:N2bXxghq3Zk+lpiOxv+m+0yjnGubVqJGJDR/9jWm3N0=
-github.com/go-zoox/dotenv v1.1.0 h1:1BiCO0S1AMnR6nVPrvznnubrURYCvCqaf+rak2ZyGRQ=
 github.com/go-zoox/dotenv v1.1.0/go.mod h1:elKALomz436YKX4Syo+f02ozIQYq2yFaprj6jL6GnCQ=
+github.com/go-zoox/dotenv v1.2.3 h1:9wx4sL2u/FrRLkzoOb7ozYii6NoGsl05KoGdZm1ebIE=
 github.com/go-zoox/encoding v1.0.5/go.mod h1:WFF+m/IWYHtClMrcKnrdZHHf3cdIF8de2uTOILPM7lw=
 github.com/go-zoox/encoding v1.0.7/go.mod h1:h6ZgI27IaUSBq1ntaZjnSruPdRFuBLNUmfp57TaPxs4=
 github.com/go-zoox/encoding v1.2.1 h1:38rQRsfL1f1YHZaqsPaGcNMkPnzatnPlYiHioUh9F4A=
@@ -94,6 +96,8 @@ github.com/go-zoox/logger v1.4.4/go.mod h1:o7ddvv/gMoMa0TomPhHoIz11ZWRbQ92pF6rwY
 github.com/go-zoox/proxy v1.2.3/go.mod h1:T+gbngAtIgvambbYibAbzrjwdu8j5pduJdA2j6RYi94=
 github.com/go-zoox/proxy v1.4.1 h1:+yUwY8bXN0wtQIXQTFD49jvRzflL/COQ4qfT5VWiuGg=
 github.com/go-zoox/proxy v1.4.1/go.mod h1:HAVpVmXgniKNMHRwgtiKNxh9YgYnduAzUFe1fpVHVjo=
+github.com/go-zoox/pubsub v1.2.2 h1:dpcFlZRSGhX0YqT/WoOJgP5bP2VDqswPqXiiFZCki1w=
+github.com/go-zoox/pubsub v1.2.2/go.mod h1:LWX0NAg80hkeGdZf7PJOEGnyN6CXooCxpIlh2MxESDo=
 github.com/go-zoox/random v1.0.0/go.mod h1:W+PTQiInxaCngiXpSvycucAKvu1tE/tKlZ9kaMp2/Ys=
 github.com/go-zoox/random v1.0.4 h1:icckpkCowQ0eGiiMkHFOJz9Qc9noOcinP+ggqWUIBH4=
 github.com/go-zoox/random v1.0.4/go.mod h1:W+PTQiInxaCngiXpSvycucAKvu1tE/tKlZ9kaMp2/Ys=
@@ -136,8 +140,8 @@ github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWm
 github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
-github.com/joho/godotenv v1.4.0 h1:3l4+N6zfMWnkbPEXKng2o2/MR5mSwTrBih4ZEkkz1lg=
 github.com/joho/godotenv v1.4.0/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
+github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
@@ -176,6 +180,8 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/redis/go-redis/v9 v9.0.5 h1:CuQcn5HIEeK7BgElubPP8CGtE0KakrnbBSTLjathl5o=
+github.com/redis/go-redis/v9 v9.0.5/go.mod h1:WqMKv5vnQbRuZstUwxQI195wHy+t4PuXDOjzMvcuQHk=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=