@@ -9,9 +9,11 @@ import (
99 "io"
1010 "net/http"
1111 "net/http/httptest"
12+ "sync"
1213 "testing"
1314 "time"
1415
16+ "github.com/cespare/xxhash/v2"
1517 "github.com/stretchr/testify/assert"
1618 "github.com/stretchr/testify/require"
1719
@@ -7996,10 +7998,70 @@ var errSubscriptionClientFail = errors.New("subscription client fail error")
79967998
79977999type FailingSubscriptionClient struct {}
79988000
7999- func (f FailingSubscriptionClient ) Subscribe (_ * resolve.Context , _ GraphQLSubscriptionOptions , _ chan <- [] byte ) error {
8001+ func (f * FailingSubscriptionClient ) Subscribe (ctx * resolve.Context , options GraphQLSubscriptionOptions , updater resolve. SubscriptionUpdater ) error {
80008002 return errSubscriptionClientFail
80018003}
80028004
8005+ func (f * FailingSubscriptionClient ) UniqueRequestID (ctx * resolve.Context , options GraphQLSubscriptionOptions , hash * xxhash.Digest ) (err error ) {
8006+ return errSubscriptionClientFail
8007+ }
8008+
8009+ type testSubscriptionUpdater struct {
8010+ updates []string
8011+ done bool
8012+ mux sync.Mutex
8013+ }
8014+
8015+ func (t * testSubscriptionUpdater ) AwaitUpdates (tt * testing.T , timeout time.Duration , count int ) {
8016+ ticker := time .NewTicker (timeout )
8017+ defer ticker .Stop ()
8018+ for {
8019+ time .Sleep (10 * time .Millisecond )
8020+ select {
8021+ case <- ticker .C :
8022+ tt .Fatalf ("timed out waiting for updates" )
8023+ default :
8024+ t .mux .Lock ()
8025+ if len (t .updates ) == count {
8026+ t .mux .Unlock ()
8027+ return
8028+ }
8029+ t .mux .Unlock ()
8030+ }
8031+ }
8032+ }
8033+
8034+ func (t * testSubscriptionUpdater ) AwaitDone (tt * testing.T , timeout time.Duration ) {
8035+ ticker := time .NewTicker (timeout )
8036+ defer ticker .Stop ()
8037+ for {
8038+ time .Sleep (10 * time .Millisecond )
8039+ select {
8040+ case <- ticker .C :
8041+ tt .Fatalf ("timed out waiting for done" )
8042+ default :
8043+ t .mux .Lock ()
8044+ if t .done {
8045+ t .mux .Unlock ()
8046+ return
8047+ }
8048+ t .mux .Unlock ()
8049+ }
8050+ }
8051+ }
8052+
8053+ func (t * testSubscriptionUpdater ) Update (data []byte ) {
8054+ t .mux .Lock ()
8055+ defer t .mux .Unlock ()
8056+ t .updates = append (t .updates , string (data ))
8057+ }
8058+
8059+ func (t * testSubscriptionUpdater ) Done () {
8060+ t .mux .Lock ()
8061+ defer t .mux .Unlock ()
8062+ t .done = true
8063+ }
8064+
80038065func TestSubscriptionSource_Start (t * testing.T ) {
80048066 chatServer := httptest .NewServer (subscriptiontesting .ChatGraphQLEndpointHandler ())
80058067 defer chatServer .Close ()
@@ -8042,84 +8104,86 @@ func TestSubscriptionSource_Start(t *testing.T) {
80428104 }
80438105
80448106 t .Run ("should return error when input is invalid" , func (t * testing.T ) {
8045- source := SubscriptionSource {client : FailingSubscriptionClient {}}
8107+ source := SubscriptionSource {client : & FailingSubscriptionClient {}}
80468108 err := source .Start (resolve .NewContext (context .Background ()), []byte (`{"url": "", "body": "", "header": null}` ), nil )
80478109 assert .Error (t , err )
80488110 })
80498111
80508112 t .Run ("should return error when subscription client returns an error" , func (t * testing.T ) {
8051- source := SubscriptionSource {client : FailingSubscriptionClient {}}
8113+ source := SubscriptionSource {client : & FailingSubscriptionClient {}}
80528114 err := source .Start (resolve .NewContext (context .Background ()), []byte (`{"url": "", "body": {}, "header": null}` ), nil )
80538115 assert .Error (t , err )
80548116 assert .Equal (t , resolve .ErrUnableToResolve , err )
80558117 })
80568118
80578119 t .Run ("invalid json: should stop before sending to upstream" , func (t * testing.T ) {
8058- next := make (chan []byte )
80598120 ctx := resolve .NewContext (context .Background ())
80608121 defer ctx .Context ().Done ()
80618122
8123+ updater := & testSubscriptionUpdater {}
8124+
80628125 source := newSubscriptionSource (ctx .Context ())
80638126 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: "#test") { text createdBy } }"}` )
8064- err := source .Start (ctx , chatSubscriptionOptions , next )
8127+ err := source .Start (ctx , chatSubscriptionOptions , updater )
80658128 require .ErrorIs (t , err , resolve .ErrUnableToResolve )
80668129 })
80678130
80688131 t .Run ("invalid syntax (roomNam)" , func (t * testing.T ) {
8069- next := make (chan []byte )
80708132 ctx := resolve .NewContext (context .Background ())
80718133 defer ctx .Context ().Done ()
80728134
8135+ updater := & testSubscriptionUpdater {}
8136+
80738137 source := newSubscriptionSource (ctx .Context ())
80748138 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomNam: \"#test\") { text createdBy } }"}` )
8075- err := source .Start (ctx , chatSubscriptionOptions , next )
8139+ err := source .Start (ctx , chatSubscriptionOptions , updater )
80768140 require .NoError (t , err )
8077-
8078- msg , ok := <- next
8079- assert .True (t , ok )
8080- 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 ))
8081- _ , ok = <- next
8082- assert .False (t , ok )
8141+ updater .AwaitUpdates (t , time .Second , 1 )
8142+ assert .Len (t , updater .updates , 1 )
8143+ 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 ])
8144+ assert .Equal (t , true , updater .done )
80838145 })
80848146
80858147 t .Run ("should close connection on stop message" , func (t * testing.T ) {
8086- next := make (chan []byte )
80878148 subscriptionLifecycle , cancelSubscription := context .WithCancel (context .Background ())
80888149 resolverLifecycle , cancelResolver := context .WithCancel (context .Background ())
80898150 defer cancelResolver ()
80908151
8152+ updater := & testSubscriptionUpdater {}
8153+
80918154 source := newSubscriptionSource (resolverLifecycle )
80928155 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}` )
8093- err := source .Start (resolve .NewContext (subscriptionLifecycle ), chatSubscriptionOptions , next )
8156+ err := source .Start (resolve .NewContext (subscriptionLifecycle ), chatSubscriptionOptions , updater )
80948157 require .NoError (t , err )
80958158
80968159 username := "myuser"
80978160 message := "hello world!"
80988161 go sendChatMessage (t , username , message )
80998162
8100- nextBytes := <- next
8101- assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , string (nextBytes ))
8163+ updater .AwaitUpdates (t , time .Second , 1 )
81028164 cancelSubscription ()
8103- _ , ok := <- next
8104- assert .False (t , ok )
8165+ updater .AwaitDone (t , time .Second * 5 )
8166+ assert .Len (t , updater .updates , 1 )
8167+ assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , updater .updates [0 ])
81058168 })
81068169
81078170 t .Run ("should successfully subscribe with chat example" , func (t * testing.T ) {
8108- next := make (chan []byte )
81098171 ctx := resolve .NewContext (context .Background ())
81108172 defer ctx .Context ().Done ()
81118173
8174+ updater := & testSubscriptionUpdater {}
8175+
81128176 source := newSubscriptionSource (ctx .Context ())
81138177 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}` )
8114- err := source .Start (ctx , chatSubscriptionOptions , next )
8178+ err := source .Start (ctx , chatSubscriptionOptions , updater )
81158179 require .NoError (t , err )
81168180
81178181 username := "myuser"
81188182 message := "hello world!"
81198183 go sendChatMessage (t , username , message )
8120-
8121- nextBytes := <- next
8122- assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , string ( nextBytes ) )
8184+ updater . AwaitUpdates ( t , time . Second , 1 )
8185+ assert . Len ( t , updater . updates , 1 )
8186+ assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , updater . updates [ 0 ] )
81238187 })
81248188}
81258189
@@ -8167,60 +8231,65 @@ func TestSubscription_GTWS_SubProtocol(t *testing.T) {
81678231 }
81688232
81698233 t .Run ("invalid syntax (roomNam)" , func (t * testing.T ) {
8170- next := make (chan []byte )
81718234 ctx := resolve .NewContext (context .Background ())
81728235 defer ctx .Context ().Done ()
81738236
8237+ updater := & testSubscriptionUpdater {}
8238+
81748239 source := newSubscriptionSource (ctx .Context ())
81758240 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomNam: \"#test\") { text createdBy } }"}` )
8176- err := source .Start (ctx , chatSubscriptionOptions , next )
8241+ err := source .Start (ctx , chatSubscriptionOptions , updater )
81778242 require .NoError (t , err )
81788243
8179- msg , ok := <- next
8180- assert .True (t , ok )
8181- 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 ) )
8182- _ , ok = <- next
8183- assert .False (t , ok )
8244+ updater . AwaitUpdates ( t , time . Second , 1 )
8245+ assert .Len (t , updater . updates , 1 )
8246+ 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 ] )
8247+ updater . AwaitDone ( t , time . Second )
8248+ assert .Equal (t , true , updater . done )
81848249 })
81858250
81868251 t .Run ("should close connection on stop message" , func (t * testing.T ) {
8187- next := make (chan []byte )
81888252 subscriptionLifecycle , cancelSubscription := context .WithCancel (context .Background ())
81898253 resolverLifecycle , cancelResolver := context .WithCancel (context .Background ())
81908254 defer cancelResolver ()
81918255
8256+ updater := & testSubscriptionUpdater {}
8257+
81928258 source := newSubscriptionSource (resolverLifecycle )
81938259 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}` )
8194- err := source .Start (resolve .NewContext (subscriptionLifecycle ), chatSubscriptionOptions , next )
8260+ err := source .Start (resolve .NewContext (subscriptionLifecycle ), chatSubscriptionOptions , updater )
81958261 require .NoError (t , err )
81968262
81978263 username := "myuser"
81988264 message := "hello world!"
81998265 go sendChatMessage (t , username , message )
82008266
8201- nextBytes := <- next
8202- assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , string (nextBytes ))
8267+ updater .AwaitUpdates (t , time .Second , 1 )
8268+ assert .Len (t , updater .updates , 1 )
8269+ assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , updater .updates [0 ])
82038270 cancelSubscription ()
8204- _ , ok := <- next
8205- assert .False (t , ok )
8271+ updater . AwaitDone ( t , time . Second * 5 )
8272+ assert .Equal (t , true , updater . done )
82068273 })
82078274
82088275 t .Run ("should successfully subscribe with chat example" , func (t * testing.T ) {
8209- next := make (chan []byte )
82108276 ctx := resolve .NewContext (context .Background ())
82118277 defer ctx .Context ().Done ()
82128278
8279+ updater := & testSubscriptionUpdater {}
8280+
82138281 source := newSubscriptionSource (ctx .Context ())
82148282 chatSubscriptionOptions := chatServerSubscriptionOptions (t , `{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}` )
8215- err := source .Start (ctx , chatSubscriptionOptions , next )
8283+ err := source .Start (ctx , chatSubscriptionOptions , updater )
82168284 require .NoError (t , err )
82178285
82188286 username := "myuser"
82198287 message := "hello world!"
82208288 go sendChatMessage (t , username , message )
82218289
8222- nextBytes := <- next
8223- assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , string (nextBytes ))
8290+ updater .AwaitUpdates (t , time .Second , 1 )
8291+ assert .Len (t , updater .updates , 1 )
8292+ assert .Equal (t , `{"data":{"messageAdded":{"text":"hello world!","createdBy":"myuser"}}}` , updater .updates [0 ])
82248293 })
82258294}
82268295
0 commit comments