mercredi 20 octobre 2021

Pubsub - How to handle connection drop from source?

Below function(BroadcastMessage) broadcast messages to subscribers(PublisherHub.subscribers). BroadcastMessage establishes a connection from source(PCF firehose)

In the below case, BroadcastMessage() establishing a connection from PCF firehose using api consumer.New()(https://github.com/cloudfoundry/noaa/blob/master/consumer/consumer.go#L78):

func BroadcastMessage(publisherHub *publisher.PublisherHub) {
    ....
    cnsmr := consumer.New(dopplerAddress, &tls.Config{InsecureSkipVerify: true}, nil)
    ....
    var (
        msgChan   <-chan *events.Envelope
        errorChan <-chan error
    )

    switch *filterType {
    case "logs":
        msgChan, errorChan = cnsmr.FilteredFirehose(firehoseSubscriptionId, authToken, consumer.LogMessages)
    case "metrics":
        msgChan, errorChan = cnsmr.FilteredFirehose(firehoseSubscriptionId, authToken, consumer.Metrics)
    default:
        msgChan, errorChan = cnsmr.Firehose(firehoseSubscriptionId, authToken)
    }

    go func() {
        for err := range errorChan {
            fmt.Fprintf(os.Stderr, "%v\n", err.Error())
        }
    }()

    for msg := range msgChan {
        publisherHub.Broadcast <- msg
    }
}

publisherHub broadcast msg to all subscribers where type PublisherHub maintains the hub of subscribers:

type PublisherHub struct {
    subscribers map[*subscriptionmediator.HandlerSubscription]struct{}
    Register    chan *subscriptionmediator.HandlerSubscription
    Unregister  chan *subscriptionmediator.HandlerSubscription
    Broadcast   chan *events.Envelope
}

main() blocks in BroadcastMessage() until explicit shutdown through SIGTERM

func main() {

    ....
    publisherHub := publisher.NewHub()
    go publisherHub.Run()
    ....
    go loadbroadcast.BroadcastMessage(publisherHub)
    ....   
}

But in production this connection(consumer.New()) is being lost due to loss of network connectivity and all the subscribers are getting affected.


How to handle loss of connection in BroadcastMessage, so that subscribers can receive messages seamlessly? Because Broadcastmessage() is already in an infinite loop(for msg := range msgChan)

Aucun commentaire:

Enregistrer un commentaire