BigBlueButton/bigbluebutton-Github
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Improve recconection flow

Browse Source
This commit is contained in:
Gustavo Trott 2024-03-13 10:08:36 -03:00
parent 1d93a1258e
commit 4ffc3578c1
9 changed files with 56 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
bbb-graphql-middleware/internal/common/SafeChannel.go
View File

@ -37,6 +37,13 @@ func (s *SafeChannel) ReceiveChannel() <-chan interface{} {
return s.ch
}
func (s *SafeChannel) Closed() bool {
s.mux.Lock()
defer s.mux.Unlock()
return s.closed
}
func (s *SafeChannel) Close() {
s.mux.Lock()
defer s.mux.Unlock()

12
bbb-graphql-middleware/internal/common/types.go
View File

@ -43,10 +43,10 @@ type BrowserConnection struct {
}
type HasuraConnection struct {
Id string // hasura connection id
Browserconn *BrowserConnection // browser connection that originated this hasura connection
Websocket *websocket.Conn // websocket used to connect to hasura
Context context.Context // hasura connection context (child of browser connection context)
ContextCancelFunc context.CancelFunc // function to cancel the hasura context (and so, the hasura connection)
MsgReceivingActiveChan *SafeChannel // indicate that it's waiting for the return of mutations before closing connection
Id string // hasura connection id
BrowserConn *BrowserConnection // browser connection that originated this hasura connection
Websocket *websocket.Conn // websocket used to connect to hasura
Context context.Context // hasura connection context (child of browser connection context)
ContextCancelFunc context.CancelFunc // function to cancel the hasura context (and so, the hasura connection)
FreezeMsgFromBrowserChan *SafeChannel // indicate that it's waiting for the return of mutations before closing connection
}

10
bbb-graphql-middleware/internal/hascli/client.go
View File

@ -60,11 +60,11 @@ func HasuraClient(browserConnection *common.BrowserConnection, cookies []*http.C
defer hasuraConnectionContextCancel()
var thisConnection = common.HasuraConnection{
Id: hasuraConnectionId,
Browserconn: browserConnection,
Context: hasuraConnectionContext,
ContextCancelFunc: hasuraConnectionContextCancel,
MsgReceivingActiveChan: common.NewSafeChannel(1),
Id: hasuraConnectionId,
BrowserConn: browserConnection,
Context: hasuraConnectionContext,
ContextCancelFunc: hasuraConnectionContextCancel,
FreezeMsgFromBrowserChan: common.NewSafeChannel(1),
}
browserConnection.HasuraConnection = &thisConnection

40
bbb-graphql-middleware/internal/hascli/conn/reader/reader.go
View File

@ -15,7 +15,7 @@ import (
// HasuraConnectionReader consumes messages from Hasura connection and add send to the browser channel
func HasuraConnectionReader(hc *common.HasuraConnection, fromHasuraToBrowserChannel *common.SafeChannel, fromBrowserToHasuraChannel *common.SafeChannel, wg *sync.WaitGroup) {
log := log.WithField("_routine", "HasuraConnectionReader").WithField("browserConnectionId", hc.Browserconn.Id).WithField("hasuraConnectionId", hc.Id)
log := log.WithField("_routine", "HasuraConnectionReader").WithField("browserConnectionId", hc.BrowserConn.Id).WithField("hasuraConnectionId", hc.Id)
defer log.Debugf("finished")
log.Debugf("starting")
@ -28,9 +28,9 @@ func HasuraConnectionReader(hc *common.HasuraConnection, fromHasuraToBrowserChan
err := wsjson.Read(hc.Context, hc.Websocket, &message)
if err != nil {
if errors.Is(err, context.Canceled) {
log.Debugf("Closing ws connection as Context was cancelled!")
log.Debugf("Closing Hasura ws connection as Context was cancelled!")
} else {
log.Errorf("Error reading message from Hasura: %v", err)
log.Debugf("Error reading message from Hasura: %v", err)
}
return
}
@ -50,9 +50,9 @@ func handleMessageReceivedFromHasura(hc *common.HasuraConnection, fromHasuraToBr
//Check if subscription is still active!
if queryId != "" {
hc.Browserconn.ActiveSubscriptionsMutex.RLock()
subscription, ok := hc.Browserconn.ActiveSubscriptions[queryId]
hc.Browserconn.ActiveSubscriptionsMutex.RUnlock()
hc.BrowserConn.ActiveSubscriptionsMutex.RLock()
subscription, ok := hc.BrowserConn.ActiveSubscriptions[queryId]
hc.BrowserConn.ActiveSubscriptionsMutex.RUnlock()
if !ok {
log.Debugf("Subscription with Id %s doesn't exist anymore, skiping response.", queryId)
return
@ -104,13 +104,13 @@ func handleSubscriptionMessage(hc *common.HasuraConnection, messageMap map[strin
//Store LastReceivedData Checksum
subscription.LastReceivedDataChecksum = dataChecksum
hc.Browserconn.ActiveSubscriptionsMutex.Lock()
hc.Browserconn.ActiveSubscriptions[queryId] = subscription
hc.Browserconn.ActiveSubscriptionsMutex.Unlock()
hc.BrowserConn.ActiveSubscriptionsMutex.Lock()
hc.BrowserConn.ActiveSubscriptions[queryId] = subscription
hc.BrowserConn.ActiveSubscriptionsMutex.Unlock()
//Apply msg patch when it supports it
if subscription.JsonPatchSupported {
msgpatch.PatchMessage(&messageMap, queryId, dataKey, dataAsJson, hc.Browserconn)
msgpatch.PatchMessage(&messageMap, queryId, dataKey, dataAsJson, hc.BrowserConn)
}
}
}
@ -126,18 +126,18 @@ func handleStreamingMessage(hc *common.HasuraConnection, messageMap map[string]i
if lastCursor != nil && subscription.StreamCursorCurrValue != lastCursor {
subscription.StreamCursorCurrValue = lastCursor
hc.Browserconn.ActiveSubscriptionsMutex.Lock()
hc.Browserconn.ActiveSubscriptions[queryId] = subscription
hc.Browserconn.ActiveSubscriptionsMutex.Unlock()
hc.BrowserConn.ActiveSubscriptionsMutex.Lock()
hc.BrowserConn.ActiveSubscriptions[queryId] = subscription
hc.BrowserConn.ActiveSubscriptionsMutex.Unlock()
}
}
func handleCompleteMessage(hc *common.HasuraConnection, queryId string) {
hc.Browserconn.ActiveSubscriptionsMutex.Lock()
queryType := hc.Browserconn.ActiveSubscriptions[queryId].Type
operationName := hc.Browserconn.ActiveSubscriptions[queryId].OperationName
delete(hc.Browserconn.ActiveSubscriptions, queryId)
hc.Browserconn.ActiveSubscriptionsMutex.Unlock()
hc.BrowserConn.ActiveSubscriptionsMutex.Lock()
queryType := hc.BrowserConn.ActiveSubscriptions[queryId].Type
operationName := hc.BrowserConn.ActiveSubscriptions[queryId].OperationName
delete(hc.BrowserConn.ActiveSubscriptions, queryId)
hc.BrowserConn.ActiveSubscriptionsMutex.Unlock()
log.Debugf("%s (%s) with Id %s finished by Hasura.", queryType, operationName, queryId)
}
@ -147,9 +147,9 @@ func handleConnectionAckMessage(hc *common.HasuraConnection, messageMap map[stri
fromBrowserToHasuraChannel.UnfreezeChannel()
//Avoid to send `connection_ack` to the browser when it's a reconnection
if hc.Browserconn.ConnAckSentToBrowser == false {
if hc.BrowserConn.ConnAckSentToBrowser == false {
fromHasuraToBrowserChannel.Send(messageMap)
hc.Browserconn.ConnAckSentToBrowser = true
hc.BrowserConn.ConnAckSentToBrowser = true
}
go retransmiter.RetransmitSubscriptionStartMessages(hc, fromBrowserToHasuraChannel)

6
bbb-graphql-middleware/internal/hascli/conn/writer/writer.go
View File

@ -14,7 +14,7 @@ import (
func HasuraConnectionWriter(hc *common.HasuraConnection, fromBrowserToHasuraChannel *common.SafeChannel, wg *sync.WaitGroup, initMessage map[string]interface{}) {
log := log.WithField("_routine", "HasuraConnectionWriter")
browserConnection := hc.Browserconn
browserConnection := hc.BrowserConn
log = log.WithField("browserConnectionId", browserConnection.Id).WithField("hasuraConnectionId", hc.Id)
@ -38,9 +38,9 @@ RangeLoop:
select {
case <-hc.Context.Done():
break RangeLoop
case <-hc.MsgReceivingActiveChan.ReceiveChannel():
case <-hc.FreezeMsgFromBrowserChan.ReceiveChannel():
if !fromBrowserToHasuraChannel.Frozen() {
log.Debugf("freezing channel fromBrowserToHasuraChannel")
log.Debug("freezing channel fromBrowserToHasuraChannel")
//Freeze channel once it's about to close Hasura connection
fromBrowserToHasuraChannel.FreezeChannel()
}

8
bbb-graphql-middleware/internal/hascli/retransmiter/retransmiter.go
View File

@ -6,10 +6,10 @@ import (
)
func RetransmitSubscriptionStartMessages(hc *common.HasuraConnection, fromBrowserToHasuraChannel *common.SafeChannel) {
log := log.WithField("_routine", "RetransmitSubscriptionStartMessages").WithField("browserConnectionId", hc.Browserconn.Id).WithField("hasuraConnectionId", hc.Id)
log := log.WithField("_routine", "RetransmitSubscriptionStartMessages").WithField("browserConnectionId", hc.BrowserConn.Id).WithField("hasuraConnectionId", hc.Id)
hc.Browserconn.ActiveSubscriptionsMutex.RLock()
for _, subscription := range hc.Browserconn.ActiveSubscriptions {
hc.BrowserConn.ActiveSubscriptionsMutex.RLock()
for _, subscription := range hc.BrowserConn.ActiveSubscriptions {
//Not retransmitting Mutations
if subscription.Type == common.Mutation {
@ -27,5 +27,5 @@ func RetransmitSubscriptionStartMessages(hc *common.HasuraConnection, fromBrowse
}
}
}
hc.Browserconn.ActiveSubscriptionsMutex.RUnlock()
hc.BrowserConn.ActiveSubscriptionsMutex.RUnlock()
}

4
bbb-graphql-middleware/internal/websrv/connhandler.go
View File

@ -138,8 +138,8 @@ func InvalidateSessionTokenConnections(sessionTokenToInvalidate string) {
for _, browserConnection := range BrowserConnections {
if browserConnection.SessionToken == sessionTokenToInvalidate {
if browserConnection.HasuraConnection != nil {
//Close chan to force stop receiving new messages from the browser
browserConnection.HasuraConnection.MsgReceivingActiveChan.Close()
//Send message to force stop receiving new messages from the browser
browserConnection.HasuraConnection.FreezeMsgFromBrowserChan.Send(true)
// Wait until there are no active mutations
for iterationCount := 0; iterationCount < 20; iterationCount++ {

7
bbb-graphql-middleware/internal/websrv/reader/reader.go
View File

@ -2,6 +2,7 @@ package reader
import (
"context"
"errors"
"github.com/iMDT/bbb-graphql-middleware/internal/common"
log "github.com/sirupsen/logrus"
"nhooyr.io/websocket"
@ -35,7 +36,11 @@ func BrowserConnectionReader(browserConnectionId string, ctx context.Context, ct
var v interface{}
err := wsjson.Read(ctx, browserWsConn, &v)
if err != nil {
log.Debugf("Browser is disconnected, skiping reading of ws message: %v", err)
if errors.Is(err, context.Canceled) {
log.Debugf("Closing Browser ws connection as Context was cancelled!")
} else {
log.Debugf("Hasura is disconnected, skiping reading of ws message: %v", err)
}
return
}

3
bbb-graphql-middleware/internal/websrv/writer/writer.go
View File

@ -24,6 +24,9 @@ RangeLoop:
case toBrowserMessage := <-fromHasuraToBrowserChannel.ReceiveChannel():
{
if toBrowserMessage == nil {
if fromHasuraToBrowserChannel.Closed() {
break RangeLoop
}
continue
}