mirror of
https://github.com/hanzoai/psrpc.git
synced 2026-09-30 23:47:22 +00:00
Run Redis and NATS in Docker during tests. (#100)
This commit is contained in:
Denys Smirnov
@@ -36,7 +36,7 @@ jobs:
|
||||
- name: Set up Go
|
||||
uses: actions/setup-go@v4
|
||||
with:
|
||||
go-version: '>=1.20'
|
||||
go-version: '>=1.22'
|
||||
|
||||
- name: Download Go modules
|
||||
run: go mod download
|
||||
@@ -50,12 +50,6 @@ jobs:
|
||||
version: latest
|
||||
args: proto
|
||||
|
||||
- name: Start Redis
|
||||
uses: supercharge/[email protected]
|
||||
|
||||
- name: Start NATS server
|
||||
run: docker run --rm --name psrpc-nats --network host -d nats:alpine
|
||||
|
||||
- name: Mage Test
|
||||
uses: magefile/mage-action@v2
|
||||
with:
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
module github.com/livekit/psrpc
|
||||
|
||||
go 1.20
|
||||
go 1.22
|
||||
|
||||
toolchain go1.23.5
|
||||
|
||||
require (
|
||||
github.com/frostbyte73/core v0.0.10
|
||||
@@ -8,9 +10,10 @@ require (
|
||||
github.com/go-logr/logr v1.3.0
|
||||
github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
|
||||
github.com/nats-io/nats.go v1.31.0
|
||||
github.com/ory/dockertest/v3 v3.11.0
|
||||
github.com/pkg/errors v0.9.1
|
||||
github.com/redis/go-redis/v9 v9.3.0
|
||||
github.com/stretchr/testify v1.8.4
|
||||
github.com/stretchr/testify v1.9.0
|
||||
github.com/twitchtv/twirp v8.1.3+incompatible
|
||||
go.uber.org/atomic v1.11.0
|
||||
go.uber.org/multierr v1.11.0
|
||||
@@ -21,18 +24,40 @@ require (
|
||||
)
|
||||
|
||||
require (
|
||||
dario.cat/mergo v1.0.0 // indirect
|
||||
github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect
|
||||
github.com/Microsoft/go-winio v0.6.2 // indirect
|
||||
github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 // indirect
|
||||
github.com/cenkalti/backoff/v4 v4.3.0 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.2.0 // indirect
|
||||
github.com/containerd/continuity v0.4.3 // indirect
|
||||
github.com/davecgh/go-spew v1.1.1 // indirect
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
|
||||
github.com/docker/cli v26.1.4+incompatible // indirect
|
||||
github.com/docker/docker v27.1.1+incompatible // indirect
|
||||
github.com/docker/go-connections v0.5.0 // indirect
|
||||
github.com/docker/go-units v0.5.0 // indirect
|
||||
github.com/gogo/protobuf v1.3.2 // indirect
|
||||
github.com/golang/protobuf v1.5.3 // indirect
|
||||
github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
|
||||
github.com/klauspost/compress v1.17.2 // indirect
|
||||
github.com/mitchellh/mapstructure v1.5.0 // indirect
|
||||
github.com/moby/docker-image-spec v1.3.1 // indirect
|
||||
github.com/moby/term v0.5.0 // indirect
|
||||
github.com/nats-io/nkeys v0.4.6 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
github.com/opencontainers/go-digest v1.0.0 // indirect
|
||||
github.com/opencontainers/image-spec v1.1.0 // indirect
|
||||
github.com/opencontainers/runc v1.1.13 // indirect
|
||||
github.com/pmezard/go-difflib v1.0.0 // indirect
|
||||
golang.org/x/crypto v0.15.0 // indirect
|
||||
golang.org/x/net v0.18.0 // indirect
|
||||
github.com/sirupsen/logrus v1.9.3 // indirect
|
||||
github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect
|
||||
github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
|
||||
github.com/xeipuuv/gojsonschema v1.2.0 // indirect
|
||||
golang.org/x/crypto v0.22.0 // indirect
|
||||
golang.org/x/sync v0.5.0 // indirect
|
||||
golang.org/x/sys v0.14.0 // indirect
|
||||
golang.org/x/sys v0.21.0 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20231106174013-bbf56f31fb17 // indirect
|
||||
gopkg.in/yaml.v2 v2.4.0 // indirect
|
||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||
)
|
||||
|
||||
@@ -1,60 +1,153 @@
|
||||
dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk=
|
||||
dario.cat/mergo v1.0.0/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
|
||||
filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
|
||||
filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
|
||||
github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 h1:L/gRVlceqvL25UVaW/CKtUDjefjrs0SPonmDGUVOYP0=
|
||||
github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E=
|
||||
github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
|
||||
github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=
|
||||
github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 h1:TngWCqHvy9oXAN6lEVMRuU21PR1EtLVZJmdB18Gu3Rw=
|
||||
github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5/go.mod h1:lmUJ/7eu/Q8D7ML55dXQrVaamCz2vxCfdQBasLZfHKk=
|
||||
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
|
||||
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
|
||||
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
|
||||
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
|
||||
github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
|
||||
github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
|
||||
github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
|
||||
github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/containerd/continuity v0.4.3 h1:6HVkalIp+2u1ZLH1J/pYX2oBVXlJZvh1X1A7bEZ9Su8=
|
||||
github.com/containerd/continuity v0.4.3/go.mod h1:F6PTNCKepoxEaXLQp3wDAjygEnImnZ/7o4JzpodfroQ=
|
||||
github.com/creack/pty v1.1.18 h1:n56/Zwd5o6whRC5PMGretI4IdRLlmBXYNjScPaBgsbY=
|
||||
github.com/creack/pty v1.1.18/go.mod h1:MOBLtS5ELjhRRrroQr9kyvTxUAFNvYEK993ew/Vr4O4=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
|
||||
github.com/docker/cli v26.1.4+incompatible h1:I8PHdc0MtxEADqYJZvhBrW9bo8gawKwwenxRM7/rLu8=
|
||||
github.com/docker/cli v26.1.4+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8=
|
||||
github.com/docker/docker v27.1.1+incompatible h1:hO/M4MtV36kzKldqnA37IWhebRA+LnqqcqDja6kVaKY=
|
||||
github.com/docker/docker v27.1.1+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
|
||||
github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj1Br63c=
|
||||
github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc=
|
||||
github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
|
||||
github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
|
||||
github.com/frostbyte73/core v0.0.10 h1:D4DQXdPb8ICayz0n75rs4UYTXrUSdxzUfeleuNJORsU=
|
||||
github.com/frostbyte73/core v0.0.10/go.mod h1:XsOGqrqe/VEV7+8vJ+3a8qnCIXNbKsoEiu/czs7nrcU=
|
||||
github.com/gammazero/deque v0.2.1 h1:qSdsbG6pgp6nL7A0+K/B7s12mcCY/5l5SIUpMOl+dC0=
|
||||
github.com/gammazero/deque v0.2.1/go.mod h1:LFroj8x4cMYCukHJDbxFCkT+r9AndaJnFMuZDV34tuU=
|
||||
github.com/go-logr/logr v1.3.0 h1:2y3SDp0ZXuc6/cjLSZ+Q3ir+QB9T/iG5yYRXqsagWSY=
|
||||
github.com/go-logr/logr v1.3.0/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
|
||||
github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y=
|
||||
github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg=
|
||||
github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
|
||||
github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
|
||||
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
|
||||
github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
|
||||
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
|
||||
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
|
||||
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
|
||||
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
|
||||
github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaUGG7oYTSPP8MxqL4YI3kZKwcP4=
|
||||
github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ=
|
||||
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
|
||||
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
|
||||
github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4=
|
||||
github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
|
||||
github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
|
||||
github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
|
||||
github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkDaKb5iXdynYrzB84ErPPO4LbRASk58=
|
||||
github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
|
||||
github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
|
||||
github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
|
||||
github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0=
|
||||
github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo=
|
||||
github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0=
|
||||
github.com/moby/term v0.5.0/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y=
|
||||
github.com/nats-io/nats.go v1.31.0 h1:/WFBHEc/dOKBF6qf1TZhrdEfTmOZ5JzdJ+Y3m6Y/p7E=
|
||||
github.com/nats-io/nats.go v1.31.0/go.mod h1:di3Bm5MLsoB4Bx61CBTsxuarI36WbhAwOm8QrW39+i8=
|
||||
github.com/nats-io/nkeys v0.4.6 h1:IzVe95ru2CT6ta874rt9saQRkWfe2nFj1NtvYSLqMzY=
|
||||
github.com/nats-io/nkeys v0.4.6/go.mod h1:4DxZNzenSVd1cYQoAa8948QY3QDjrHfcfVADymtkpts=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||
github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
|
||||
github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
|
||||
github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug=
|
||||
github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM=
|
||||
github.com/opencontainers/runc v1.1.13 h1:98S2srgG9vw0zWcDpFMn5TRrh8kLxa/5OFUstuUhmRs=
|
||||
github.com/opencontainers/runc v1.1.13/go.mod h1:R016aXacfp/gwQBYw2FDGa9m+n6atbLWrYY8hNMT/sA=
|
||||
github.com/ory/dockertest/v3 v3.11.0 h1:OiHcxKAvSDUwsEVh2BjxQQc/5EHz9n0va9awCtNGuyA=
|
||||
github.com/ory/dockertest/v3 v3.11.0/go.mod h1:VIPxS1gwT9NpPOrfD3rACs8Y9Z7yhzO4SB194iUDnUI=
|
||||
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.3.0 h1:RiVDjmig62jIWp7Kk4XVLs0hzV6pI3PyTnnL0cnn0u0=
|
||||
github.com/redis/go-redis/v9 v9.3.0/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
|
||||
github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
|
||||
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
|
||||
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
|
||||
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
|
||||
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
|
||||
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
|
||||
github.com/twitchtv/twirp v8.1.3+incompatible h1:+F4TdErPgSUbMZMwp13Q/KgDVuI7HJXP61mNV3/7iuU=
|
||||
github.com/twitchtv/twirp v8.1.3+incompatible/go.mod h1:RRJoFSAmTEh2weEqWtpPE3vFK5YBhA6bqp2l1kfCC5A=
|
||||
github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
|
||||
github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb h1:zGWFAtiMcyryUHoUjUJX0/lt1H2+i2Ka2n+D3DImSNo=
|
||||
github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
|
||||
github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 h1:EzJWgHovont7NscjpAxXsDA8S8BMYve8Y5+7cuRE7R0=
|
||||
github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
|
||||
github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17UxZ74=
|
||||
github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
|
||||
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
|
||||
go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
|
||||
go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
|
||||
go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
|
||||
golang.org/x/crypto v0.15.0 h1:frVn1TEaCEaZcn3Tmd7Y2b5KKPaZ+I32Q2OA3kYp5TA=
|
||||
golang.org/x/crypto v0.15.0/go.mod h1:4ChreQoLWfG3xLDer1WdlH5NdlQ3+mwnQq1YTKY+72g=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/crypto v0.22.0 h1:g1v0xeRhjcugydODzvb3mEM9SQ0HGp9s/nh3COQ/C30=
|
||||
golang.org/x/crypto v0.22.0/go.mod h1:vr6Su+7cTlO45qkww3VDJlzDn0ctJvRgYbC2NvXHt+M=
|
||||
golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa h1:FRnLl4eNAQl8hwxVVC17teOw8kdjVDVAiFMtgUdTSRQ=
|
||||
golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa/go.mod h1:zk2irFbV9DP96SEBUUAy67IdHUaZuSnrz1n472HUCLE=
|
||||
golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
||||
golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
||||
golang.org/x/mod v0.14.0 h1:dGoOF9QVLYng8IHTm7BAyWqCqSheQ5pYWGhzW00YJr0=
|
||||
golang.org/x/mod v0.14.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
|
||||
golang.org/x/net v0.18.0 h1:mIYleuAkSbHh0tCv7RvjL3F6ZVbLjq4+R7zbOn3Kokg=
|
||||
golang.org/x/net v0.18.0/go.mod h1:/czyP5RqHAH4odGYxBJ1qz0+CE5WZ+2j1YgoEo8F2jQ=
|
||||
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
|
||||
golang.org/x/net v0.24.0 h1:1PcaxkF854Fu3+lvBIx5SYn9wRlBzzcnHZSiaFFAb0w=
|
||||
golang.org/x/net v0.24.0/go.mod h1:2Q7sJY5mzlzWjKtYUEXSlBWCdyaioyXzRB2RtU8KVE8=
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.5.0 h1:60k92dhOjHxJkrqnwsfl8KuaHbn/5dl0lUPUklKo3qE=
|
||||
golang.org/x/sync v0.5.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
|
||||
golang.org/x/sys v0.14.0 h1:Vz7Qs629MkJkGyHxUlRHizWJRG2j8fbQKjELVSNhy7Q=
|
||||
golang.org/x/sys v0.14.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws=
|
||||
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
|
||||
golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
|
||||
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
|
||||
golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
|
||||
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20231106174013-bbf56f31fb17 h1:Jyp0Hsi0bmHXG6k9eATXoYtjd6e2UzZ1SCn/wIupY14=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20231106174013-bbf56f31fb17/go.mod h1:oQ5rr10WTTMvP4A36n8JpR1OrO1BEiV4f78CneXZxkA=
|
||||
google.golang.org/grpc v1.59.0 h1:Z5Iec2pjwb+LEOqzpB2MR12/eKFhDPhuqW91O+4bwUk=
|
||||
@@ -65,5 +158,10 @@ google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs
|
||||
google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
|
||||
gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gotest.tools/v3 v3.5.1 h1:EENdUnS3pdur5nybKYIh2Vfgc8IUNBjxDPSjtiJcOzU=
|
||||
gotest.tools/v3 v3.5.1/go.mod h1:isy3WKz7GK6uNw/sbHzfKBLvlvXwUyV06n6brMxxopU=
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package bus
|
||||
package bus_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
@@ -21,27 +21,24 @@ import (
|
||||
"time"
|
||||
"unsafe"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/atomic"
|
||||
"google.golang.org/protobuf/types/known/wrapperspb"
|
||||
|
||||
"github.com/livekit/psrpc/internal/bus"
|
||||
"github.com/livekit/psrpc/internal/bus/bustest"
|
||||
)
|
||||
|
||||
func redisTestChannel(channel string) Channel {
|
||||
return Channel{Legacy: channel}
|
||||
func redisTestChannel(channel string) bus.Channel {
|
||||
return bus.Channel{Legacy: channel}
|
||||
}
|
||||
|
||||
func TestRedisMessageBus(t *testing.T) {
|
||||
t.Run("published messages are received by subscribers", func(t *testing.T) {
|
||||
rc0 := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
rc1 := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
t.Cleanup(func() {
|
||||
rc0.Close()
|
||||
rc1.Close()
|
||||
})
|
||||
srv := bustest.NewRedis(t, bustest.Docker(t))
|
||||
|
||||
b0 := NewRedisMessageBus(rc0)
|
||||
b1 := NewRedisMessageBus(rc1)
|
||||
t.Run("published messages are received by subscribers", func(t *testing.T) {
|
||||
b0 := srv.Connect(t)
|
||||
b1 := srv.Connect(t)
|
||||
|
||||
r, err := b0.Subscribe(context.Background(), redisTestChannel("test"), 100)
|
||||
require.NoError(t, err)
|
||||
@@ -53,27 +50,18 @@ func TestRedisMessageBus(t *testing.T) {
|
||||
err = b1.Publish(context.Background(), redisTestChannel("test"), src)
|
||||
require.NoError(t, err)
|
||||
|
||||
b, ok := r.read()
|
||||
b, ok := bus.RawRead(r)
|
||||
require.True(t, ok)
|
||||
|
||||
dst, err := deserialize(b)
|
||||
dst, err := bus.Deserialize(b)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, src.Value, dst.(*wrapperspb.StringValue).Value)
|
||||
})
|
||||
|
||||
t.Run("published messages are received by only one queue subscriber", func(t *testing.T) {
|
||||
rc0 := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
rc1 := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
rc2 := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
t.Cleanup(func() {
|
||||
rc0.Close()
|
||||
rc1.Close()
|
||||
rc2.Close()
|
||||
})
|
||||
|
||||
b0 := NewRedisMessageBus(rc0)
|
||||
b1 := NewRedisMessageBus(rc1)
|
||||
b2 := NewRedisMessageBus(rc2)
|
||||
b0 := srv.Connect(t)
|
||||
b1 := srv.Connect(t)
|
||||
b2 := srv.Connect(t)
|
||||
|
||||
r1, err := b1.SubscribeQueue(context.Background(), redisTestChannel("test"), 100)
|
||||
require.NoError(t, err)
|
||||
@@ -90,12 +78,12 @@ func TestRedisMessageBus(t *testing.T) {
|
||||
var n atomic.Int64
|
||||
|
||||
go func() {
|
||||
if _, ok := r1.read(); ok {
|
||||
if _, ok := bus.RawRead(r1); ok {
|
||||
n.Inc()
|
||||
}
|
||||
}()
|
||||
go func() {
|
||||
if _, ok := r2.read(); ok {
|
||||
if _, ok := bus.RawRead(r2); ok {
|
||||
n.Inc()
|
||||
}
|
||||
}()
|
||||
@@ -106,18 +94,9 @@ func TestRedisMessageBus(t *testing.T) {
|
||||
})
|
||||
|
||||
t.Run("closed subscriptions are unreadable", func(t *testing.T) {
|
||||
rc0 := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
rc1 := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
rc2 := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
t.Cleanup(func() {
|
||||
rc0.Close()
|
||||
rc1.Close()
|
||||
rc2.Close()
|
||||
})
|
||||
|
||||
b0 := NewRedisMessageBus(rc0)
|
||||
b1 := NewRedisMessageBus(rc1)
|
||||
b2 := NewRedisMessageBus(rc2)
|
||||
b0 := srv.Connect(t)
|
||||
b1 := srv.Connect(t)
|
||||
b2 := srv.Connect(t)
|
||||
|
||||
r1, err := b1.Subscribe(context.Background(), redisTestChannel("test"), 100)
|
||||
require.NoError(t, err)
|
||||
@@ -131,9 +110,9 @@ func TestRedisMessageBus(t *testing.T) {
|
||||
err = b0.Publish(context.Background(), redisTestChannel("test"), src)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, ok := r1.read()
|
||||
_, ok := bus.RawRead(r1)
|
||||
require.True(t, ok)
|
||||
_, ok = r2.read()
|
||||
_, ok = bus.RawRead(r2)
|
||||
require.True(t, ok)
|
||||
|
||||
err = r1.Close()
|
||||
@@ -144,23 +123,18 @@ func TestRedisMessageBus(t *testing.T) {
|
||||
err = b0.Publish(context.Background(), redisTestChannel("test"), src)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, ok = r1.read()
|
||||
_, ok = bus.RawRead(r1)
|
||||
require.False(t, ok)
|
||||
_, ok = r2.read()
|
||||
_, ok = bus.RawRead(r2)
|
||||
require.True(t, ok)
|
||||
})
|
||||
}
|
||||
|
||||
func BenchmarkRedisMessageBus(b *testing.B) {
|
||||
rc0 := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
rc1 := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
b.Cleanup(func() {
|
||||
rc0.Close()
|
||||
rc1.Close()
|
||||
})
|
||||
srv := bustest.NewRedis(b, bustest.Docker(b))
|
||||
|
||||
b0 := NewRedisMessageBus(rc0)
|
||||
b1 := NewRedisMessageBus(rc1)
|
||||
b0 := srv.Connect(b)
|
||||
b1 := srv.Connect(b)
|
||||
|
||||
r, _ := b0.Subscribe(context.Background(), redisTestChannel("test"), 100)
|
||||
|
||||
@@ -169,7 +143,7 @@ func BenchmarkRedisMessageBus(b *testing.B) {
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
for i := 0; i < b.N; i++ {
|
||||
r.read()
|
||||
bus.RawRead(r)
|
||||
}
|
||||
close(done)
|
||||
}()
|
||||
|
||||
+20
-36
@@ -12,66 +12,50 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package bus
|
||||
package bus_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/livekit/psrpc/internal"
|
||||
"github.com/livekit/psrpc/internal/bus"
|
||||
"github.com/livekit/psrpc/internal/bus/bustest"
|
||||
"github.com/livekit/psrpc/pkg/rand"
|
||||
)
|
||||
|
||||
const defaultClientTimeout = time.Second * 3
|
||||
|
||||
func busTestChannel(channel string) Channel {
|
||||
return Channel{
|
||||
func busTestChannel(channel string) bus.Channel {
|
||||
return bus.Channel{
|
||||
Legacy: channel,
|
||||
Server: channel,
|
||||
}
|
||||
}
|
||||
|
||||
func TestMessageBus(t *testing.T) {
|
||||
t.Run("Local", func(t *testing.T) {
|
||||
bus := NewLocalMessageBus()
|
||||
t.Run("testSubscribe", func(t *testing.T) { testSubscribe(t, bus) })
|
||||
t.Run("testSubscribeQueue", func(t *testing.T) { testSubscribeQueue(t, bus) })
|
||||
t.Run("testSubscribeClose", func(t *testing.T) { testSubscribeClose(t, bus) })
|
||||
})
|
||||
|
||||
t.Run("Redis", func(t *testing.T) {
|
||||
rc := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
bus := NewRedisMessageBus(rc)
|
||||
t.Run("testSubscribe", func(t *testing.T) { testSubscribe(t, bus) })
|
||||
t.Run("testSubscribeQueue", func(t *testing.T) { testSubscribeQueue(t, bus) })
|
||||
t.Run("testSubscribeClose", func(t *testing.T) { testSubscribeClose(t, bus) })
|
||||
})
|
||||
|
||||
t.Run("Nats", func(t *testing.T) {
|
||||
nc, _ := nats.Connect(nats.DefaultURL)
|
||||
bus := NewNatsMessageBus(nc)
|
||||
t.Run("testSubscribe", func(t *testing.T) { testSubscribe(t, bus) })
|
||||
t.Run("testSubscribeQueue", func(t *testing.T) { testSubscribeQueue(t, bus) })
|
||||
t.Run("testSubscribeClose", func(t *testing.T) { testSubscribeClose(t, bus) })
|
||||
bustest.TestAll(t, func(t *testing.T, bus func(t testing.TB) bus.MessageBus) {
|
||||
b := bus(t)
|
||||
t.Run("testSubscribe", func(t *testing.T) { testSubscribe(t, b) })
|
||||
t.Run("testSubscribeQueue", func(t *testing.T) { testSubscribeQueue(t, b) })
|
||||
t.Run("testSubscribeClose", func(t *testing.T) { testSubscribeClose(t, b) })
|
||||
})
|
||||
}
|
||||
|
||||
func testSubscribe(t *testing.T, bus MessageBus) {
|
||||
func testSubscribe(t *testing.T, b bus.MessageBus) {
|
||||
ctx := context.Background()
|
||||
|
||||
channel := rand.NewString()
|
||||
subA, err := Subscribe[*internal.Request](ctx, bus, busTestChannel(channel), DefaultChannelSize)
|
||||
subA, err := bus.Subscribe[*internal.Request](ctx, b, busTestChannel(channel), bus.DefaultChannelSize)
|
||||
require.NoError(t, err)
|
||||
subB, err := Subscribe[*internal.Request](ctx, bus, busTestChannel(channel), DefaultChannelSize)
|
||||
subB, err := bus.Subscribe[*internal.Request](ctx, b, busTestChannel(channel), bus.DefaultChannelSize)
|
||||
require.NoError(t, err)
|
||||
time.Sleep(time.Millisecond * 100)
|
||||
|
||||
require.NoError(t, bus.Publish(ctx, busTestChannel(channel), &internal.Request{
|
||||
require.NoError(t, b.Publish(ctx, busTestChannel(channel), &internal.Request{
|
||||
RequestId: "1",
|
||||
}))
|
||||
|
||||
@@ -83,17 +67,17 @@ func testSubscribe(t *testing.T, bus MessageBus) {
|
||||
require.Equal(t, "1", msgB.RequestId)
|
||||
}
|
||||
|
||||
func testSubscribeQueue(t *testing.T, bus MessageBus) {
|
||||
func testSubscribeQueue(t *testing.T, b bus.MessageBus) {
|
||||
ctx := context.Background()
|
||||
|
||||
channel := rand.NewString()
|
||||
subA, err := SubscribeQueue[*internal.Request](ctx, bus, busTestChannel(channel), DefaultChannelSize)
|
||||
subA, err := bus.SubscribeQueue[*internal.Request](ctx, b, busTestChannel(channel), bus.DefaultChannelSize)
|
||||
require.NoError(t, err)
|
||||
subB, err := SubscribeQueue[*internal.Request](ctx, bus, busTestChannel(channel), DefaultChannelSize)
|
||||
subB, err := bus.SubscribeQueue[*internal.Request](ctx, b, busTestChannel(channel), bus.DefaultChannelSize)
|
||||
require.NoError(t, err)
|
||||
time.Sleep(time.Millisecond * 100)
|
||||
|
||||
require.NoError(t, bus.Publish(ctx, busTestChannel(channel), &internal.Request{
|
||||
require.NoError(t, b.Publish(ctx, busTestChannel(channel), &internal.Request{
|
||||
RequestId: "2",
|
||||
}))
|
||||
|
||||
@@ -119,11 +103,11 @@ func testSubscribeQueue(t *testing.T, bus MessageBus) {
|
||||
require.Equal(t, 1, received)
|
||||
}
|
||||
|
||||
func testSubscribeClose(t *testing.T, bus MessageBus) {
|
||||
func testSubscribeClose(t *testing.T, b bus.MessageBus) {
|
||||
ctx := context.Background()
|
||||
|
||||
channel := rand.NewString()
|
||||
sub, err := Subscribe[*internal.Request](ctx, bus, busTestChannel(channel), DefaultChannelSize)
|
||||
sub, err := bus.Subscribe[*internal.Request](ctx, b, busTestChannel(channel), bus.DefaultChannelSize)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NoError(t, sub.Close())
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
// Copyright 2025 LiveKit, Inc.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package bustest
|
||||
|
||||
import (
|
||||
"math/rand/v2"
|
||||
"net"
|
||||
"testing"
|
||||
|
||||
"github.com/ory/dockertest/v3"
|
||||
|
||||
"github.com/livekit/psrpc/internal/bus"
|
||||
)
|
||||
|
||||
var (
|
||||
baseID = rand.Uint32N(1000)
|
||||
servers []serverInfo
|
||||
)
|
||||
|
||||
type serverInfo struct {
|
||||
Name string
|
||||
Func ServerFunc
|
||||
}
|
||||
|
||||
type ServerFunc func(t testing.TB, pool *dockertest.Pool) Server
|
||||
|
||||
func RegisterServer(name string, fnc ServerFunc) {
|
||||
servers = append(servers, serverInfo{
|
||||
Name: name,
|
||||
Func: fnc,
|
||||
})
|
||||
}
|
||||
|
||||
func waitTCPPort(t testing.TB, pool *dockertest.Pool, addr string) {
|
||||
if err := pool.Retry(func() error {
|
||||
conn, err := net.Dial("tcp", addr)
|
||||
if err != nil {
|
||||
t.Log(err)
|
||||
return err
|
||||
}
|
||||
_ = conn.Close()
|
||||
return nil
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func Docker(t testing.TB) *dockertest.Pool {
|
||||
pool, err := dockertest.NewPool("")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = pool.Client.Ping()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return pool
|
||||
}
|
||||
|
||||
type Server interface {
|
||||
Connect(t testing.TB) bus.MessageBus
|
||||
}
|
||||
|
||||
func TestAll(t *testing.T, test func(t *testing.T, bus func(t testing.TB) bus.MessageBus)) {
|
||||
pool := Docker(t)
|
||||
for _, c := range servers {
|
||||
c := c
|
||||
t.Run(c.Name, func(t *testing.T) {
|
||||
s := c.Func(t, pool)
|
||||
test(t, s.Connect)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
package bustest
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/ory/dockertest/v3"
|
||||
|
||||
"github.com/livekit/psrpc/internal/bus"
|
||||
)
|
||||
|
||||
func init() {
|
||||
RegisterServer("Local", func(t testing.TB, pool *dockertest.Pool) Server {
|
||||
return NewLocalBus()
|
||||
})
|
||||
}
|
||||
|
||||
func NewLocalBus() Server {
|
||||
b := bus.NewLocalMessageBus()
|
||||
return &localBus{b: b}
|
||||
}
|
||||
|
||||
type localBus struct {
|
||||
b bus.MessageBus
|
||||
}
|
||||
|
||||
func (s *localBus) Connect(t testing.TB) bus.MessageBus {
|
||||
return s.b
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
package bustest
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/ory/dockertest/v3"
|
||||
|
||||
"github.com/livekit/psrpc/internal/bus"
|
||||
)
|
||||
|
||||
func init() {
|
||||
RegisterServer("NATS", NewNATS)
|
||||
}
|
||||
|
||||
var natsLast = baseID
|
||||
|
||||
func NewNATS(t testing.TB, pool *dockertest.Pool) Server {
|
||||
c, err := pool.RunWithOptions(&dockertest.RunOptions{
|
||||
Name: fmt.Sprintf("psrpc-nats-%d", atomic.AddUint32(&natsLast, 1)),
|
||||
Repository: "nats", Tag: "latest",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
_ = pool.Purge(c)
|
||||
})
|
||||
addr := c.GetHostPort("4222/tcp")
|
||||
waitTCPPort(t, pool, addr)
|
||||
|
||||
t.Log("NATS running on", addr)
|
||||
|
||||
s := &natsServer{addr: "nats://" + addr}
|
||||
|
||||
err = pool.Retry(func() error {
|
||||
nc, err := s.connect()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
nc.Close()
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
type natsServer struct {
|
||||
addr string
|
||||
}
|
||||
|
||||
func (s *natsServer) connect() (*nats.Conn, error) {
|
||||
nc, err := nats.Connect(s.addr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := nc.Flush(); err != nil {
|
||||
nc.Close()
|
||||
return nil, err
|
||||
}
|
||||
return nc, nil
|
||||
}
|
||||
|
||||
func (s *natsServer) Connect(t testing.TB) bus.MessageBus {
|
||||
nc, err := s.connect()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return bus.NewNatsMessageBus(nc)
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
package bustest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/ory/dockertest/v3"
|
||||
"github.com/redis/go-redis/v9"
|
||||
|
||||
"github.com/livekit/psrpc/internal/bus"
|
||||
)
|
||||
|
||||
func init() {
|
||||
RegisterServer("Redis", NewRedis)
|
||||
}
|
||||
|
||||
var redisLast = baseID
|
||||
|
||||
func NewRedis(t testing.TB, pool *dockertest.Pool) Server {
|
||||
c, err := pool.RunWithOptions(&dockertest.RunOptions{
|
||||
Name: fmt.Sprintf("psrpc-redis-%d", atomic.AddUint32(&redisLast, 1)),
|
||||
Repository: "redis", Tag: "latest",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
_ = pool.Purge(c)
|
||||
})
|
||||
addr := c.GetHostPort("6379/tcp")
|
||||
waitTCPPort(t, pool, addr)
|
||||
|
||||
t.Log("Redis running on", addr)
|
||||
|
||||
s := &redisServer{addr: addr}
|
||||
|
||||
err = pool.Retry(func() error {
|
||||
rc, err := s.connect()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_ = rc.Close()
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
return s
|
||||
}
|
||||
|
||||
type redisServer struct {
|
||||
addr string
|
||||
}
|
||||
|
||||
func (s *redisServer) connect() (redis.UniversalClient, error) {
|
||||
rc := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{s.addr}})
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
|
||||
if err := rc.Ping(ctx).Err(); err != nil {
|
||||
_ = rc.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return rc, nil
|
||||
}
|
||||
|
||||
func (s *redisServer) Connect(t testing.TB) bus.MessageBus {
|
||||
rc, err := s.connect()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return bus.NewRedisMessageBus(rc)
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
package bus
|
||||
|
||||
import "google.golang.org/protobuf/proto"
|
||||
|
||||
// These helpers are only exposed during tests.
|
||||
|
||||
func RawRead(r Reader) ([]byte, bool) {
|
||||
return r.read()
|
||||
}
|
||||
|
||||
func Deserialize(b []byte) (proto.Message, error) {
|
||||
return deserialize(b)
|
||||
}
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: empty_service.proto
|
||||
|
||||
@@ -44,7 +44,7 @@ var file_empty_service_proto_rawDesc = []byte{
|
||||
0x63, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
|
||||
}
|
||||
|
||||
var file_empty_service_proto_goTypes = []interface{}{}
|
||||
var file_empty_service_proto_goTypes = []any{}
|
||||
var file_empty_service_proto_depIdxs = []int32{
|
||||
0, // [0:0] is the sub-list for method output_type
|
||||
0, // [0:0] is the sub-list for method input_type
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: service.proto
|
||||
|
||||
@@ -54,7 +54,7 @@ var file_service_proto_rawDesc = []byte{
|
||||
0x6f, 0x74, 0x6f, 0x33,
|
||||
}
|
||||
|
||||
var file_service_proto_goTypes = []interface{}{
|
||||
var file_service_proto_goTypes = []any{
|
||||
(*wrapperspb.StringValue)(nil), // 0: google.protobuf.StringValue
|
||||
(*emptypb.Empty)(nil), // 1: google.protobuf.Empty
|
||||
}
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: importable.proto
|
||||
|
||||
@@ -37,18 +37,16 @@ const (
|
||||
)
|
||||
|
||||
type Msg struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *Msg) Reset() {
|
||||
*x = Msg{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_importable_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_importable_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *Msg) String() string {
|
||||
@@ -59,7 +57,7 @@ func (*Msg) ProtoMessage() {}
|
||||
|
||||
func (x *Msg) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_importable_proto_msgTypes[0]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -105,7 +103,7 @@ func file_importable_proto_rawDescGZIP() []byte {
|
||||
}
|
||||
|
||||
var file_importable_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
|
||||
var file_importable_proto_goTypes = []interface{}{
|
||||
var file_importable_proto_goTypes = []any{
|
||||
(*Msg)(nil), // 0: psrpc.internal.test.importable.Msg
|
||||
}
|
||||
var file_importable_proto_depIdxs = []int32{
|
||||
@@ -123,20 +121,6 @@ func file_importable_proto_init() {
|
||||
if File_importable_proto != nil {
|
||||
return
|
||||
}
|
||||
if !protoimpl.UnsafeEnabled {
|
||||
file_importable_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*Msg); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: importer.proto
|
||||
|
||||
@@ -55,7 +55,7 @@ var file_importer_proto_rawDesc = []byte{
|
||||
0x72, 0x6f, 0x74, 0x6f, 0x33,
|
||||
}
|
||||
|
||||
var file_importer_proto_goTypes = []interface{}{
|
||||
var file_importer_proto_goTypes = []any{
|
||||
(*importable.Msg)(nil), // 0: psrpc.internal.test.importable.Msg
|
||||
}
|
||||
var file_importer_proto_depIdxs = []int32{
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: importer_local.proto
|
||||
|
||||
@@ -54,7 +54,7 @@ var file_importer_local_proto_rawDesc = []byte{
|
||||
0x6c, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
|
||||
}
|
||||
|
||||
var file_importer_local_proto_goTypes = []interface{}{
|
||||
var file_importer_local_proto_goTypes = []any{
|
||||
(*Msg)(nil), // 0: psrpc.internal.test.importer_local.Msg
|
||||
}
|
||||
var file_importer_local_proto_depIdxs = []int32{
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: importer_local_msgdef.proto
|
||||
|
||||
@@ -35,18 +35,16 @@ const (
|
||||
)
|
||||
|
||||
type Msg struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *Msg) Reset() {
|
||||
*x = Msg{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_importer_local_msgdef_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_importer_local_msgdef_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *Msg) String() string {
|
||||
@@ -57,7 +55,7 @@ func (*Msg) ProtoMessage() {}
|
||||
|
||||
func (x *Msg) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_importer_local_msgdef_proto_msgTypes[0]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -99,7 +97,7 @@ func file_importer_local_msgdef_proto_rawDescGZIP() []byte {
|
||||
}
|
||||
|
||||
var file_importer_local_msgdef_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
|
||||
var file_importer_local_msgdef_proto_goTypes = []interface{}{
|
||||
var file_importer_local_msgdef_proto_goTypes = []any{
|
||||
(*Msg)(nil), // 0: psrpc.internal.test.importer_local.Msg
|
||||
}
|
||||
var file_importer_local_msgdef_proto_depIdxs = []int32{
|
||||
@@ -115,20 +113,6 @@ func file_importer_local_msgdef_proto_init() {
|
||||
if File_importer_local_msgdef_proto != nil {
|
||||
return
|
||||
}
|
||||
if !protoimpl.UnsafeEnabled {
|
||||
file_importer_local_msgdef_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*Msg); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: x/x.proto
|
||||
|
||||
@@ -52,7 +52,7 @@ var file_x_x_proto_rawDesc = []byte{
|
||||
0x2f, 0x78, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
|
||||
}
|
||||
|
||||
var file_x_x_proto_goTypes = []interface{}{
|
||||
var file_x_x_proto_goTypes = []any{
|
||||
(*y.MsgY)(nil), // 0: psrpc.internal.test.importmapping.y.MsgY
|
||||
}
|
||||
var file_x_x_proto_depIdxs = []int32{
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: y/y.proto
|
||||
|
||||
@@ -35,18 +35,16 @@ const (
|
||||
)
|
||||
|
||||
type MsgY struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *MsgY) Reset() {
|
||||
*x = MsgY{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_y_y_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_y_y_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *MsgY) String() string {
|
||||
@@ -57,7 +55,7 @@ func (*MsgY) ProtoMessage() {}
|
||||
|
||||
func (x *MsgY) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_y_y_proto_msgTypes[0]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -97,7 +95,7 @@ func file_y_y_proto_rawDescGZIP() []byte {
|
||||
}
|
||||
|
||||
var file_y_y_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
|
||||
var file_y_y_proto_goTypes = []interface{}{
|
||||
var file_y_y_proto_goTypes = []any{
|
||||
(*MsgY)(nil), // 0: psrpc.internal.test.importmapping.y.MsgY
|
||||
}
|
||||
var file_y_y_proto_depIdxs = []int32{
|
||||
@@ -113,20 +111,6 @@ func file_y_y_proto_init() {
|
||||
if File_y_y_proto != nil {
|
||||
return
|
||||
}
|
||||
if !protoimpl.UnsafeEnabled {
|
||||
file_y_y_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*MsgY); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: multiple1.proto
|
||||
|
||||
@@ -37,18 +37,16 @@ const (
|
||||
)
|
||||
|
||||
type Msg1 struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *Msg1) Reset() {
|
||||
*x = Msg1{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_multiple1_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_multiple1_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *Msg1) String() string {
|
||||
@@ -59,7 +57,7 @@ func (*Msg1) ProtoMessage() {}
|
||||
|
||||
func (x *Msg1) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_multiple1_proto_msgTypes[0]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -103,7 +101,7 @@ func file_multiple1_proto_rawDescGZIP() []byte {
|
||||
}
|
||||
|
||||
var file_multiple1_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
|
||||
var file_multiple1_proto_goTypes = []interface{}{
|
||||
var file_multiple1_proto_goTypes = []any{
|
||||
(*Msg1)(nil), // 0: psrpc.internal.test.multiple.Msg1
|
||||
}
|
||||
var file_multiple1_proto_depIdxs = []int32{
|
||||
@@ -121,20 +119,6 @@ func file_multiple1_proto_init() {
|
||||
if File_multiple1_proto != nil {
|
||||
return
|
||||
}
|
||||
if !protoimpl.UnsafeEnabled {
|
||||
file_multiple1_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*Msg1); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: multiple2.proto
|
||||
|
||||
@@ -37,18 +37,16 @@ const (
|
||||
)
|
||||
|
||||
type Msg2 struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *Msg2) Reset() {
|
||||
*x = Msg2{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_multiple2_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_multiple2_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *Msg2) String() string {
|
||||
@@ -59,7 +57,7 @@ func (*Msg2) ProtoMessage() {}
|
||||
|
||||
func (x *Msg2) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_multiple2_proto_msgTypes[0]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -110,7 +108,7 @@ func file_multiple2_proto_rawDescGZIP() []byte {
|
||||
}
|
||||
|
||||
var file_multiple2_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
|
||||
var file_multiple2_proto_goTypes = []interface{}{
|
||||
var file_multiple2_proto_goTypes = []any{
|
||||
(*Msg2)(nil), // 0: psrpc.internal.test.multiple.Msg2
|
||||
(*Msg1)(nil), // 1: psrpc.internal.test.multiple.Msg1
|
||||
}
|
||||
@@ -132,20 +130,6 @@ func file_multiple2_proto_init() {
|
||||
return
|
||||
}
|
||||
file_multiple1_proto_init()
|
||||
if !protoimpl.UnsafeEnabled {
|
||||
file_multiple2_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*Msg2); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: my_service.proto
|
||||
|
||||
@@ -36,18 +36,16 @@ const (
|
||||
)
|
||||
|
||||
type Ignored struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *Ignored) Reset() {
|
||||
*x = Ignored{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_my_service_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_my_service_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *Ignored) String() string {
|
||||
@@ -58,7 +56,7 @@ func (*Ignored) ProtoMessage() {}
|
||||
|
||||
func (x *Ignored) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_my_service_proto_msgTypes[0]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -74,18 +72,16 @@ func (*Ignored) Descriptor() ([]byte, []int) {
|
||||
}
|
||||
|
||||
type MyRequest struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *MyRequest) Reset() {
|
||||
*x = MyRequest{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_my_service_proto_msgTypes[1]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_my_service_proto_msgTypes[1]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *MyRequest) String() string {
|
||||
@@ -96,7 +92,7 @@ func (*MyRequest) ProtoMessage() {}
|
||||
|
||||
func (x *MyRequest) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_my_service_proto_msgTypes[1]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -112,18 +108,16 @@ func (*MyRequest) Descriptor() ([]byte, []int) {
|
||||
}
|
||||
|
||||
type MyResponse struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *MyResponse) Reset() {
|
||||
*x = MyResponse{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_my_service_proto_msgTypes[2]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_my_service_proto_msgTypes[2]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *MyResponse) String() string {
|
||||
@@ -134,7 +128,7 @@ func (*MyResponse) ProtoMessage() {}
|
||||
|
||||
func (x *MyResponse) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_my_service_proto_msgTypes[2]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -150,18 +144,16 @@ func (*MyResponse) Descriptor() ([]byte, []int) {
|
||||
}
|
||||
|
||||
type MyUpdate struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *MyUpdate) Reset() {
|
||||
*x = MyUpdate{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_my_service_proto_msgTypes[3]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_my_service_proto_msgTypes[3]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *MyUpdate) String() string {
|
||||
@@ -172,7 +164,7 @@ func (*MyUpdate) ProtoMessage() {}
|
||||
|
||||
func (x *MyUpdate) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_my_service_proto_msgTypes[3]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -188,18 +180,16 @@ func (*MyUpdate) Descriptor() ([]byte, []int) {
|
||||
}
|
||||
|
||||
type MyClientMessage struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *MyClientMessage) Reset() {
|
||||
*x = MyClientMessage{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_my_service_proto_msgTypes[4]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_my_service_proto_msgTypes[4]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *MyClientMessage) String() string {
|
||||
@@ -210,7 +200,7 @@ func (*MyClientMessage) ProtoMessage() {}
|
||||
|
||||
func (x *MyClientMessage) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_my_service_proto_msgTypes[4]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -226,18 +216,16 @@ func (*MyClientMessage) Descriptor() ([]byte, []int) {
|
||||
}
|
||||
|
||||
type MyServerMessage struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *MyServerMessage) Reset() {
|
||||
*x = MyServerMessage{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_my_service_proto_msgTypes[5]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_my_service_proto_msgTypes[5]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *MyServerMessage) String() string {
|
||||
@@ -248,7 +236,7 @@ func (*MyServerMessage) ProtoMessage() {}
|
||||
|
||||
func (x *MyServerMessage) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_my_service_proto_msgTypes[5]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -345,7 +333,7 @@ func file_my_service_proto_rawDescGZIP() []byte {
|
||||
}
|
||||
|
||||
var file_my_service_proto_msgTypes = make([]protoimpl.MessageInfo, 6)
|
||||
var file_my_service_proto_goTypes = []interface{}{
|
||||
var file_my_service_proto_goTypes = []any{
|
||||
(*Ignored)(nil), // 0: psrpc.internal.test.customservice.Ignored
|
||||
(*MyRequest)(nil), // 1: psrpc.internal.test.customservice.MyRequest
|
||||
(*MyResponse)(nil), // 2: psrpc.internal.test.customservice.MyResponse
|
||||
@@ -380,80 +368,6 @@ func file_my_service_proto_init() {
|
||||
if File_my_service_proto != nil {
|
||||
return
|
||||
}
|
||||
if !protoimpl.UnsafeEnabled {
|
||||
file_my_service_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*Ignored); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
file_my_service_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*MyRequest); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
file_my_service_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*MyResponse); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
file_my_service_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*MyUpdate); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
file_my_service_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*MyClientMessage); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
file_my_service_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*MyServerMessage); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
|
||||
@@ -21,43 +21,24 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"github.com/stretchr/testify/require"
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
"github.com/livekit/psrpc"
|
||||
"github.com/livekit/psrpc/internal/bus"
|
||||
"github.com/livekit/psrpc/internal/bus/bustest"
|
||||
)
|
||||
|
||||
func TestGeneratedService(t *testing.T) {
|
||||
t.Run("Local", func(t *testing.T) {
|
||||
testGeneratedService(t, (func() func() psrpc.MessageBus {
|
||||
bus := psrpc.NewLocalMessageBus()
|
||||
return func() psrpc.MessageBus { return bus }
|
||||
})())
|
||||
})
|
||||
|
||||
t.Run("Redis", func(t *testing.T) {
|
||||
testGeneratedService(t, func() psrpc.MessageBus {
|
||||
rc := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
return psrpc.NewRedisMessageBus(rc)
|
||||
})
|
||||
})
|
||||
|
||||
t.Run("Nats", func(t *testing.T) {
|
||||
testGeneratedService(t, func() psrpc.MessageBus {
|
||||
nc, _ := nats.Connect(nats.DefaultURL)
|
||||
return psrpc.NewNatsMessageBus(nc)
|
||||
})
|
||||
})
|
||||
bustest.TestAll(t, testGeneratedService)
|
||||
}
|
||||
|
||||
func testGeneratedService(t *testing.T, bus func() psrpc.MessageBus) {
|
||||
func testGeneratedService(t *testing.T, bus func(t testing.TB) bus.MessageBus) {
|
||||
ctx := context.Background()
|
||||
req := &MyRequest{}
|
||||
update := &MyUpdate{}
|
||||
sA := createServer(t, bus())
|
||||
sB := createServer(t, bus())
|
||||
sA := createServer(t, bus(t))
|
||||
sB := createServer(t, bus(t))
|
||||
|
||||
t.Cleanup(func() {
|
||||
shutdown(t, sA)
|
||||
@@ -72,8 +53,8 @@ func testGeneratedService(t *testing.T, bus func() psrpc.MessageBus) {
|
||||
responseHook := func(ctx context.Context, req proto.Message, rpcInfo psrpc.RPCInfo, res proto.Message, err error) {
|
||||
responseCount++
|
||||
}
|
||||
cA := createClient(t, bus(), psrpc.WithClientRequestHooks(requestHook), psrpc.WithClientResponseHooks(responseHook))
|
||||
cB := createClient(t, bus())
|
||||
cA := createClient(t, bus(t), psrpc.WithClientRequestHooks(requestHook), psrpc.WithClientResponseHooks(responseHook))
|
||||
cB := createClient(t, bus(t))
|
||||
|
||||
// rpc NormalRPC(MyRequest) returns (MyResponse);
|
||||
_, err := cA.NormalRPC(ctx, req)
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: no_package_name.proto
|
||||
|
||||
@@ -35,18 +35,16 @@ const (
|
||||
)
|
||||
|
||||
type Msg struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *Msg) Reset() {
|
||||
*x = Msg{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_no_package_name_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_no_package_name_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *Msg) String() string {
|
||||
@@ -57,7 +55,7 @@ func (*Msg) ProtoMessage() {}
|
||||
|
||||
func (x *Msg) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_no_package_name_proto_msgTypes[0]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -98,7 +96,7 @@ func file_no_package_name_proto_rawDescGZIP() []byte {
|
||||
}
|
||||
|
||||
var file_no_package_name_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
|
||||
var file_no_package_name_proto_goTypes = []interface{}{
|
||||
var file_no_package_name_proto_goTypes = []any{
|
||||
(*Msg)(nil), // 0: Msg
|
||||
}
|
||||
var file_no_package_name_proto_depIdxs = []int32{
|
||||
@@ -116,20 +114,6 @@ func file_no_package_name_proto_init() {
|
||||
if File_no_package_name_proto != nil {
|
||||
return
|
||||
}
|
||||
if !protoimpl.UnsafeEnabled {
|
||||
file_no_package_name_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*Msg); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: no_package_name_importer/no_package_name_importer.proto
|
||||
|
||||
@@ -52,7 +52,7 @@ var file_no_package_name_importer_no_package_name_importer_proto_rawDesc = []byt
|
||||
0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
|
||||
}
|
||||
|
||||
var file_no_package_name_importer_no_package_name_importer_proto_goTypes = []interface{}{
|
||||
var file_no_package_name_importer_no_package_name_importer_proto_goTypes = []any{
|
||||
(*no_package_name.Msg)(nil), // 0: Msg
|
||||
}
|
||||
var file_no_package_name_importer_no_package_name_importer_proto_depIdxs = []int32{
|
||||
|
||||
+16
-45
@@ -16,17 +16,16 @@ package test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/livekit/psrpc"
|
||||
"github.com/livekit/psrpc/internal"
|
||||
"github.com/livekit/psrpc/internal/bus"
|
||||
"github.com/livekit/psrpc/internal/bus/bustest"
|
||||
"github.com/livekit/psrpc/pkg/client"
|
||||
"github.com/livekit/psrpc/pkg/info"
|
||||
"github.com/livekit/psrpc/pkg/rand"
|
||||
@@ -34,59 +33,31 @@ import (
|
||||
)
|
||||
|
||||
func TestRPC(t *testing.T) {
|
||||
cases := []struct {
|
||||
label string
|
||||
bus func() psrpc.MessageBus
|
||||
}{
|
||||
{
|
||||
label: "Local",
|
||||
bus: (func() func() psrpc.MessageBus {
|
||||
bus := psrpc.NewLocalMessageBus()
|
||||
return func() psrpc.MessageBus { return bus }
|
||||
})(),
|
||||
},
|
||||
{
|
||||
label: "Redis",
|
||||
bus: func() psrpc.MessageBus {
|
||||
rc := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{"localhost:6379"}})
|
||||
return psrpc.NewRedisMessageBus(rc)
|
||||
},
|
||||
},
|
||||
{
|
||||
label: "Nats",
|
||||
bus: func() psrpc.MessageBus {
|
||||
nc, _ := nats.Connect(nats.DefaultURL)
|
||||
return psrpc.NewNatsMessageBus(nc)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, c := range cases {
|
||||
c := c
|
||||
t.Run(fmt.Sprintf("RPC/%s", c.label), func(t *testing.T) {
|
||||
testRPC(t, c.bus)
|
||||
bustest.TestAll(t, func(t *testing.T, bus func(t testing.TB) bus.MessageBus) {
|
||||
t.Run("RPC", func(t *testing.T) {
|
||||
testRPC(t, bus)
|
||||
})
|
||||
t.Run(fmt.Sprintf("Stream/%s", c.label), func(t *testing.T) {
|
||||
testStream(t, c.bus)
|
||||
t.Run("Stream", func(t *testing.T) {
|
||||
testStream(t, bus)
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func testRPC(t *testing.T, bus func() psrpc.MessageBus) {
|
||||
func testRPC(t *testing.T, bus func(t testing.TB) bus.MessageBus) {
|
||||
serviceName := "test"
|
||||
|
||||
serverA := server.NewRPCServer(&info.ServiceDefinition{
|
||||
Name: serviceName,
|
||||
ID: rand.NewString(),
|
||||
}, bus())
|
||||
}, bus(t))
|
||||
serverB := server.NewRPCServer(&info.ServiceDefinition{
|
||||
Name: serviceName,
|
||||
ID: rand.NewString(),
|
||||
}, bus())
|
||||
}, bus(t))
|
||||
serverC := server.NewRPCServer(&info.ServiceDefinition{
|
||||
Name: serviceName,
|
||||
ID: rand.NewString(),
|
||||
}, bus())
|
||||
}, bus(t))
|
||||
|
||||
t.Cleanup(func() {
|
||||
serverA.Close(true)
|
||||
@@ -97,7 +68,7 @@ func testRPC(t *testing.T, bus func() psrpc.MessageBus) {
|
||||
c, err := client.NewRPCClient(&info.ServiceDefinition{
|
||||
Name: serviceName,
|
||||
ID: rand.NewString(),
|
||||
}, bus())
|
||||
}, bus(t))
|
||||
require.NoError(t, err)
|
||||
|
||||
retErr := psrpc.NewErrorf(psrpc.Internal, "foo")
|
||||
@@ -173,13 +144,13 @@ func testRPC(t *testing.T, bus func() psrpc.MessageBus) {
|
||||
}
|
||||
}
|
||||
|
||||
func testStream(t *testing.T, bus func() psrpc.MessageBus) {
|
||||
func testStream(t *testing.T, bus func(t testing.TB) bus.MessageBus) {
|
||||
serviceName := "test_stream"
|
||||
|
||||
serverA := server.NewRPCServer(&info.ServiceDefinition{
|
||||
Name: serviceName,
|
||||
ID: rand.NewString(),
|
||||
}, bus())
|
||||
}, bus(t))
|
||||
|
||||
t.Cleanup(func() {
|
||||
serverA.Close(true)
|
||||
@@ -188,7 +159,7 @@ func testStream(t *testing.T, bus func() psrpc.MessageBus) {
|
||||
c, err := client.NewRPCClientWithStreams(&info.ServiceDefinition{
|
||||
Name: serviceName,
|
||||
ID: rand.NewString(),
|
||||
}, bus())
|
||||
}, bus(t))
|
||||
require.NoError(t, err)
|
||||
|
||||
serverClose := make(chan struct{})
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: service_method_same_name.proto
|
||||
|
||||
@@ -35,18 +35,16 @@ const (
|
||||
)
|
||||
|
||||
type Msg struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *Msg) Reset() {
|
||||
*x = Msg{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_service_method_same_name_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_service_method_same_name_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *Msg) String() string {
|
||||
@@ -57,7 +55,7 @@ func (*Msg) ProtoMessage() {}
|
||||
|
||||
func (x *Msg) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_service_method_same_name_proto_msgTypes[0]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -97,7 +95,7 @@ func file_service_method_same_name_proto_rawDescGZIP() []byte {
|
||||
}
|
||||
|
||||
var file_service_method_same_name_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
|
||||
var file_service_method_same_name_proto_goTypes = []interface{}{
|
||||
var file_service_method_same_name_proto_goTypes = []any{
|
||||
(*Msg)(nil), // 0: Msg
|
||||
}
|
||||
var file_service_method_same_name_proto_depIdxs = []int32{
|
||||
@@ -115,20 +113,6 @@ func file_service_method_same_name_proto_init() {
|
||||
if File_service_method_same_name_proto != nil {
|
||||
return
|
||||
}
|
||||
if !protoimpl.UnsafeEnabled {
|
||||
file_service_method_same_name_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*Msg); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.32.0
|
||||
// protoc-gen-go v1.36.3
|
||||
// protoc v4.23.4
|
||||
// source: snake_case_names.proto
|
||||
|
||||
@@ -38,18 +38,16 @@ const (
|
||||
)
|
||||
|
||||
type MakeHatArgsV1 struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *MakeHatArgsV1) Reset() {
|
||||
*x = MakeHatArgsV1{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_snake_case_names_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_snake_case_names_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *MakeHatArgsV1) String() string {
|
||||
@@ -60,7 +58,7 @@ func (*MakeHatArgsV1) ProtoMessage() {}
|
||||
|
||||
func (x *MakeHatArgsV1) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_snake_case_names_proto_msgTypes[0]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -76,22 +74,19 @@ func (*MakeHatArgsV1) Descriptor() ([]byte, []int) {
|
||||
}
|
||||
|
||||
type MakeHatArgsV1_HatV1 struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Size int32 `protobuf:"varint,1,opt,name=size,proto3" json:"size,omitempty"`
|
||||
Color string `protobuf:"bytes,2,opt,name=color,proto3" json:"color,omitempty"`
|
||||
Name string `protobuf:"bytes,3,opt,name=name,proto3" json:"name,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
|
||||
Size int32 `protobuf:"varint,1,opt,name=size,proto3" json:"size,omitempty"`
|
||||
Color string `protobuf:"bytes,2,opt,name=color,proto3" json:"color,omitempty"`
|
||||
Name string `protobuf:"bytes,3,opt,name=name,proto3" json:"name,omitempty"`
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *MakeHatArgsV1_HatV1) Reset() {
|
||||
*x = MakeHatArgsV1_HatV1{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_snake_case_names_proto_msgTypes[1]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_snake_case_names_proto_msgTypes[1]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *MakeHatArgsV1_HatV1) String() string {
|
||||
@@ -102,7 +97,7 @@ func (*MakeHatArgsV1_HatV1) ProtoMessage() {}
|
||||
|
||||
func (x *MakeHatArgsV1_HatV1) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_snake_case_names_proto_msgTypes[1]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -139,20 +134,17 @@ func (x *MakeHatArgsV1_HatV1) GetName() string {
|
||||
}
|
||||
|
||||
type MakeHatArgsV1_SizeV1 struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Inches int32 `protobuf:"varint,1,opt,name=inches,proto3" json:"inches,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
|
||||
Inches int32 `protobuf:"varint,1,opt,name=inches,proto3" json:"inches,omitempty"`
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *MakeHatArgsV1_SizeV1) Reset() {
|
||||
*x = MakeHatArgsV1_SizeV1{}
|
||||
if protoimpl.UnsafeEnabled {
|
||||
mi := &file_snake_case_names_proto_msgTypes[2]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
mi := &file_snake_case_names_proto_msgTypes[2]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *MakeHatArgsV1_SizeV1) String() string {
|
||||
@@ -163,7 +155,7 @@ func (*MakeHatArgsV1_SizeV1) ProtoMessage() {}
|
||||
|
||||
func (x *MakeHatArgsV1_SizeV1) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_snake_case_names_proto_msgTypes[2]
|
||||
if protoimpl.UnsafeEnabled && x != nil {
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
@@ -227,7 +219,7 @@ func file_snake_case_names_proto_rawDescGZIP() []byte {
|
||||
}
|
||||
|
||||
var file_snake_case_names_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
|
||||
var file_snake_case_names_proto_goTypes = []interface{}{
|
||||
var file_snake_case_names_proto_goTypes = []any{
|
||||
(*MakeHatArgsV1)(nil), // 0: psrpc.internal.test.snake_case_names.MakeHatArgs_v1
|
||||
(*MakeHatArgsV1_HatV1)(nil), // 1: psrpc.internal.test.snake_case_names.MakeHatArgs_v1.Hat_v1
|
||||
(*MakeHatArgsV1_SizeV1)(nil), // 2: psrpc.internal.test.snake_case_names.MakeHatArgs_v1.Size_v1
|
||||
@@ -247,44 +239,6 @@ func file_snake_case_names_proto_init() {
|
||||
if File_snake_case_names_proto != nil {
|
||||
return
|
||||
}
|
||||
if !protoimpl.UnsafeEnabled {
|
||||
file_snake_case_names_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*MakeHatArgsV1); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
file_snake_case_names_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*MakeHatArgsV1_HatV1); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
file_snake_case_names_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
|
||||
switch v := v.(*MakeHatArgsV1_SizeV1); i {
|
||||
case 0:
|
||||
return &v.state
|
||||
case 1:
|
||||
return &v.sizeCache
|
||||
case 2:
|
||||
return &v.unknownFields
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
|
||||
Reference in New Issue
Block a user