@@ -9,11 +9,9 @@ import (
99 "io"
1010 "net/http"
1111 "net/http/httptest"
12- "sync"
1312 "testing"
1413 "time"
1514
16- "github.com/cespare/xxhash/v2"
1715 "github.com/stretchr/testify/assert"
1816 "github.com/stretchr/testify/require"
1917
@@ -8076,70 +8074,10 @@ var errSubscriptionClientFail = errors.New("subscription client fail error")
80768074
80778075type FailingSubscriptionClient struct {}
80788076
8079- func (f * FailingSubscriptionClient ) Subscribe (ctx * resolve.Context , options GraphQLSubscriptionOptions , updater resolve. SubscriptionUpdater ) error {
8077+ func (f FailingSubscriptionClient ) Subscribe (_ * resolve.Context , _ GraphQLSubscriptionOptions , _ chan <- [] byte ) error {
80808078 return errSubscriptionClientFail
80818079}
80828080
8083- func (f * FailingSubscriptionClient ) UniqueRequestID (ctx * resolve.Context , options GraphQLSubscriptionOptions , hash * xxhash.Digest ) (err error ) {
8084- return errSubscriptionClientFail
8085- }
8086-
8087- type testSubscriptionUpdater struct {
8088- updates []string
8089- done bool
8090- mux sync.Mutex
8091- }
8092-
8093- func (t * testSubscriptionUpdater ) AwaitUpdates (tt * testing.T , timeout time.Duration , count int ) {
8094- ticker := time .NewTicker (timeout )
8095- defer ticker .Stop ()
8096- for {
8097- time .Sleep (10 * time .Millisecond )
8098- select {
8099- case <- ticker .C :
8100- tt .Fatalf ("timed out waiting for updates" )
8101- default :
8102- t .mux .Lock ()
8103- if len (t .updates ) == count {
8104- t .mux .Unlock ()
8105- return
8106- }
8107- t .mux .Unlock ()
8108- }
8109- }
8110- }
8111-
8112- func (t * testSubscriptionUpdater ) AwaitDone (tt * testing.T , timeout time.Duration ) {
8113- ticker := time .NewTicker (timeout )
8114- defer ticker .Stop ()
8115- for {
8116- time .Sleep (10 * time .Millisecond )
8117- select {
8118- case <- ticker .C :
8119- tt .Fatalf ("timed out waiting for done" )
8120- default :
8121- t .mux .Lock ()
8122- if t .done {
8123- t .mux .Unlock ()
8124- return
8125- }
8126- t .mux .Unlock ()
8127- }
8128- }
8129- }
8130-
8131- func (t * testSubscriptionUpdater ) Update (data []byte ) {
8132- t .mux .Lock ()
8133- defer t .mux .Unlock ()
8134- t .updates = append (t .updates , string (data ))
8135- }
8136-
8137- func (t * testSubscriptionUpdater ) Done () {
8138- t .mux .Lock ()
8139- defer t .mux .Unlock ()
8140- t .done = true
8141- }
8142-
81438081func TestSubscriptionSource_Start (t * testing.T ) {
81448082 chatServer := httptest .NewServer (subscriptiontesting .ChatGraphQLEndpointHandler ())
81458083 defer chatServer .Close ()
@@ -8182,86 +8120,84 @@ func TestSubscriptionSource_Start(t *testing.T) {
81828120 }
81838121
81848122 t .Run ("should return error when input is invalid" , func (t * testing.T ) {
8185- source := SubscriptionSource {client : & FailingSubscriptionClient {}}
8123+ source := SubscriptionSource {client : FailingSubscriptionClient {}}
81868124 err := source .Start (resolve .NewContext (context .Background ()), []byte (`{"url": "", "body": "", "header": null}` ), nil )
81878125 assert .Error (t , err )
81888126 })
81898127
81908128 t .Run ("should return error when subscription client returns an error" , func (t * testing.T ) {
8191- source := SubscriptionSource {client : & FailingSubscriptionClient {}}
8129+ source := SubscriptionSource {client : FailingSubscriptionClient {}}
81928130 err := source .Start (resolve .NewContext (context .Background ()), []byte (`{"url": "", "body": {}, "header": null}` ), nil )
81938131 assert .Error (t , err )
81948132 assert .Equal (t , resolve .ErrUnableToResolve , err )
81958133 })
81968134
81978135 t .Run ("invalid json: should stop before sending to upstream" , func (t * testing.T ) {
8136+ next := make (chan []byte )
81988137 ctx := resolve .NewContext (context .Background ())
81998138 defer ctx .Context ().Done ()
82008139
8201- updater := & testSubscriptionUpdater {}
8202-
82038140 source := newSubscriptionSource (ctx .Context ())
82048141 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: "#test") { text createdBy } }"}` )
8205- err := source .Start (ctx , chatSubscriptionOptions , updater )
8142+ err := source .Start (ctx , chatSubscriptionOptions , next )
82068143 require .ErrorIs (t , err , resolve .ErrUnableToResolve )
82078144 })
82088145
82098146 t .Run ("invalid syntax (roomNam)" , func (t * testing.T ) {
8147+ next := make (chan []byte )
82108148 ctx := resolve .NewContext (context .Background ())
82118149 defer ctx .Context ().Done ()
82128150
8213- updater := & testSubscriptionUpdater {}
8214-
82158151 source := newSubscriptionSource (ctx .Context ())
82168152 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomNam: \"#test\") { text createdBy } }"}` )
8217- err := source .Start (ctx , chatSubscriptionOptions , updater )
8153+ err := source .Start (ctx , chatSubscriptionOptions , next )
82188154 require .NoError (t , err )
8219- updater .AwaitUpdates (t , time .Second , 1 )
8220- assert .Len (t , updater .updates , 1 )
8221- assert .Equal (t , `{"errors":[{"message":"Unknown argument \"roomNam\" on field \"Subscription.messageAdded\". Did you mean \"roomName\"?","locations":[{"line":1,"column":29}],"extensions":{"code":"GRAPHQL_VALIDATION_FAILED"}},{"message":"Field \"messageAdded\" argument \"roomName\" of type \"String!\" is required, but it was not provided.","locations":[{"line":1,"column":29}],"extensions":{"code":"GRAPHQL_VALIDATION_FAILED"}}]}` , updater .updates [0 ])
8222- assert .Equal (t , true , updater .done )
8155+
8156+ msg , ok := <- next
8157+ assert .True (t , ok )
8158+ assert .Equal (t , `{"errors":[{"message":"Unknown argument \"roomNam\" on field \"Subscription.messageAdded\". Did you mean \"roomName\"?","locations":[{"line":1,"column":29}],"extensions":{"code":"GRAPHQL_VALIDATION_FAILED"}},{"message":"Field \"messageAdded\" argument \"roomName\" of type \"String!\" is required, but it was not provided.","locations":[{"line":1,"column":29}],"extensions":{"code":"GRAPHQL_VALIDATION_FAILED"}}]}` , string (msg ))
8159+ _ , ok = <- next
8160+ assert .False (t , ok )
82238161 })
82248162
82258163 t .Run ("should close connection on stop message" , func (t * testing.T ) {
8164+ next := make (chan []byte )
82268165 subscriptionLifecycle , cancelSubscription := context .WithCancel (context .Background ())
82278166 resolverLifecycle , cancelResolver := context .WithCancel (context .Background ())
82288167 defer cancelResolver ()
82298168
8230- updater := & testSubscriptionUpdater {}
8231-
82328169 source := newSubscriptionSource (resolverLifecycle )
82338170 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}` )
8234- err := source .Start (resolve .NewContext (subscriptionLifecycle ), chatSubscriptionOptions , updater )
8171+ err := source .Start (resolve .NewContext (subscriptionLifecycle ), chatSubscriptionOptions , next )
82358172 require .NoError (t , err )
82368173
82378174 username := "myuser"
82388175 message := "hello world!"
82398176 go sendChatMessage (t , username , message )
82408177
8241- updater .AwaitUpdates (t , time .Second , 1 )
8178+ nextBytes := <- next
8179+ assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , string (nextBytes ))
82428180 cancelSubscription ()
8243- updater .AwaitDone (t , time .Second * 5 )
8244- assert .Len (t , updater .updates , 1 )
8245- assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , updater .updates [0 ])
8181+ _ , ok := <- next
8182+ assert .False (t , ok )
82468183 })
82478184
82488185 t .Run ("should successfully subscribe with chat example" , func (t * testing.T ) {
8186+ next := make (chan []byte )
82498187 ctx := resolve .NewContext (context .Background ())
82508188 defer ctx .Context ().Done ()
82518189
8252- updater := & testSubscriptionUpdater {}
8253-
82548190 source := newSubscriptionSource (ctx .Context ())
82558191 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}` )
8256- err := source .Start (ctx , chatSubscriptionOptions , updater )
8192+ err := source .Start (ctx , chatSubscriptionOptions , next )
82578193 require .NoError (t , err )
82588194
82598195 username := "myuser"
82608196 message := "hello world!"
82618197 go sendChatMessage (t , username , message )
8262- updater . AwaitUpdates ( t , time . Second , 1 )
8263- assert . Len ( t , updater . updates , 1 )
8264- assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , updater . updates [ 0 ] )
8198+
8199+ nextBytes := <- next
8200+ assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , string ( nextBytes ) )
82658201 })
82668202}
82678203
@@ -8309,65 +8245,60 @@ func TestSubscription_GTWS_SubProtocol(t *testing.T) {
83098245 }
83108246
83118247 t .Run ("invalid syntax (roomNam)" , func (t * testing.T ) {
8248+ next := make (chan []byte )
83128249 ctx := resolve .NewContext (context .Background ())
83138250 defer ctx .Context ().Done ()
83148251
8315- updater := & testSubscriptionUpdater {}
8316-
83178252 source := newSubscriptionSource (ctx .Context ())
83188253 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomNam: \"#test\") { text createdBy } }"}` )
8319- err := source .Start (ctx , chatSubscriptionOptions , updater )
8254+ err := source .Start (ctx , chatSubscriptionOptions , next )
83208255 require .NoError (t , err )
83218256
8322- updater . AwaitUpdates ( t , time . Second , 1 )
8323- assert .Len (t , updater . updates , 1 )
8324- assert .Equal (t , `{"errors":[{"message":"Unknown argument \"roomNam\" on field \"Subscription.messageAdded\". Did you mean \"roomName\"?","locations":[{"line":1,"column":29}],"extensions":{"code":"GRAPHQL_VALIDATION_FAILED"}},{"message":"Field \"messageAdded\" argument \"roomName\" of type \"String!\" is required, but it was not provided.","locations":[{"line":1,"column":29}],"extensions":{"code":"GRAPHQL_VALIDATION_FAILED"}}]}` , updater . updates [ 0 ] )
8325- updater . AwaitDone ( t , time . Second )
8326- assert .Equal (t , true , updater . done )
8257+ msg , ok := <- next
8258+ assert .True (t , ok )
8259+ assert .Equal (t , `{"errors":[{"message":"Unknown argument \"roomNam\" on field \"Subscription.messageAdded\". Did you mean \"roomName\"?","locations":[{"line":1,"column":29}],"extensions":{"code":"GRAPHQL_VALIDATION_FAILED"}},{"message":"Field \"messageAdded\" argument \"roomName\" of type \"String!\" is required, but it was not provided.","locations":[{"line":1,"column":29}],"extensions":{"code":"GRAPHQL_VALIDATION_FAILED"}}]}` , string ( msg ) )
8260+ _ , ok = <- next
8261+ assert .False (t , ok )
83278262 })
83288263
83298264 t .Run ("should close connection on stop message" , func (t * testing.T ) {
8265+ next := make (chan []byte )
83308266 subscriptionLifecycle , cancelSubscription := context .WithCancel (context .Background ())
83318267 resolverLifecycle , cancelResolver := context .WithCancel (context .Background ())
83328268 defer cancelResolver ()
83338269
8334- updater := & testSubscriptionUpdater {}
8335-
83368270 source := newSubscriptionSource (resolverLifecycle )
83378271 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}` )
8338- err := source .Start (resolve .NewContext (subscriptionLifecycle ), chatSubscriptionOptions , updater )
8272+ err := source .Start (resolve .NewContext (subscriptionLifecycle ), chatSubscriptionOptions , next )
83398273 require .NoError (t , err )
83408274
83418275 username := "myuser"
83428276 message := "hello world!"
83438277 go sendChatMessage (t , username , message )
83448278
8345- updater .AwaitUpdates (t , time .Second , 1 )
8346- assert .Len (t , updater .updates , 1 )
8347- assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , updater .updates [0 ])
8279+ nextBytes := <- next
8280+ assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , string (nextBytes ))
83488281 cancelSubscription ()
8349- updater . AwaitDone ( t , time . Second * 5 )
8350- assert .Equal (t , true , updater . done )
8282+ _ , ok := <- next
8283+ assert .False (t , ok )
83518284 })
83528285
83538286 t .Run ("should successfully subscribe with chat example" , func (t * testing.T ) {
8287+ next := make (chan []byte )
83548288 ctx := resolve .NewContext (context .Background ())
83558289 defer ctx .Context ().Done ()
83568290
8357- updater := & testSubscriptionUpdater {}
8358-
83598291 source := newSubscriptionSource (ctx .Context ())
83608292 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}` )
8361- err := source .Start (ctx , chatSubscriptionOptions , updater )
8293+ err := source .Start (ctx , chatSubscriptionOptions , next )
83628294 require .NoError (t , err )
83638295
83648296 username := "myuser"
83658297 message := "hello world!"
83668298 go sendChatMessage (t , username , message )
83678299
8368- updater .AwaitUpdates (t , time .Second , 1 )
8369- assert .Len (t , updater .updates , 1 )
8370- assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , updater .updates [0 ])
8300+ nextBytes := <- next
8301+ assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , string (nextBytes ))
83718302 })
83728303}
83738304
0 commit comments