Browse Source

able to detect disconnected subscribers

pull/1318/head
Chris Lu 5 years ago
parent
commit
b0de01ff3b
  1. 19
      weed/messaging/broker/broker_grpc_server_subscribe.go
  2. 6
      weed/messaging/msgclient/sub_chan.go
  3. 40
      weed/messaging/msgclient/subscriber.go

19
weed/messaging/broker/broker_grpc_server_subscribe.go

@ -50,6 +50,17 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs
lock := broker.topicManager.RequestLock(tp, topicConfig, false) lock := broker.topicManager.RequestLock(tp, topicConfig, false)
defer broker.topicManager.ReleaseLock(tp, false) defer broker.topicManager.ReleaseLock(tp, false)
isConnected := true
go func() {
for isConnected {
time.Sleep(1737 * time.Millisecond)
if err = stream.Send(&messaging_pb.BrokerMessage{}); err != nil {
isConnected = false
lock.cond.Signal()
}
}
}()
lastReadTime := time.Now() lastReadTime := time.Now()
switch in.Init.StartPosition { switch in.Init.StartPosition {
case messaging_pb.SubscriberMessage_InitMessage_TIMESTAMP: case messaging_pb.SubscriberMessage_InitMessage_TIMESTAMP:
@ -93,8 +104,10 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs
} }
if err := broker.readPersistedLogBuffer(&tp, lastReadTime, eachLogEntryFn); err != nil { if err := broker.readPersistedLogBuffer(&tp, lastReadTime, eachLogEntryFn); err != nil {
// println("stopping from persisted logs")
return err
if err != io.EOF {
println("stopping from persisted logs", err.Error())
return err
}
} }
if processedTsNs != 0 { if processedTsNs != 0 {
@ -105,7 +118,7 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs
lock.Mutex.Lock() lock.Mutex.Lock()
lock.cond.Wait() lock.cond.Wait()
lock.Mutex.Unlock() lock.Mutex.Unlock()
return true
return isConnected
}, eachLogEntryFn) }, eachLogEntryFn)
return err return err

6
weed/messaging/msgclient/sub_chan.go

@ -27,7 +27,7 @@ func (mc *MessagingClient) NewSubChannel(subscriberId, chanName string) (*SubCha
if err != nil { if err != nil {
return nil, err return nil, err
} }
sc, err := setupSubscriberClient(grpcConnection, subscriberId, "chan", chanName, 0, time.Unix(0, 0))
sc, err := setupSubscriberClient(grpcConnection, tp, subscriberId, time.Unix(0, 0))
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -48,6 +48,10 @@ func (mc *MessagingClient) NewSubChannel(subscriberId, chanName string) (*SubCha
log.Printf("fail to receive from netchan %s: %v", chanName, subErr) log.Printf("fail to receive from netchan %s: %v", chanName, subErr)
return return
} }
if resp.Data == nil {
// this could be heartbeat from broker
continue
}
if resp.Data.IsClose { if resp.Data.IsClose {
t.stream.Send(&messaging_pb.SubscriberMessage{ t.stream.Send(&messaging_pb.SubscriberMessage{
IsClose: true, IsClose: true,

40
weed/messaging/msgclient/subscriber.go

@ -5,6 +5,7 @@ import (
"io" "io"
"time" "time"
"github.com/chrislusf/seaweedfs/weed/messaging/broker"
"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb" "github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
"google.golang.org/grpc" "google.golang.org/grpc"
) )
@ -14,7 +15,6 @@ type Subscriber struct {
subscriberId string subscriberId string
} }
/*
func (mc *MessagingClient) NewSubscriber(subscriberId, namespace, topic string, startTime time.Time) (*Subscriber, error) { func (mc *MessagingClient) NewSubscriber(subscriberId, namespace, topic string, startTime time.Time) (*Subscriber, error) {
// read topic configuration // read topic configuration
topicConfiguration := &messaging_pb.TopicConfiguration{ topicConfiguration := &messaging_pb.TopicConfiguration{
@ -23,7 +23,16 @@ func (mc *MessagingClient) NewSubscriber(subscriberId, namespace, topic string,
subscriberClients := make([]messaging_pb.SeaweedMessaging_SubscribeClient, topicConfiguration.PartitionCount) subscriberClients := make([]messaging_pb.SeaweedMessaging_SubscribeClient, topicConfiguration.PartitionCount)
for i := 0; i < int(topicConfiguration.PartitionCount); i++ { for i := 0; i < int(topicConfiguration.PartitionCount); i++ {
client, err := mc.setupSubscriberClient(subscriberId, namespace, topic, int32(i), startTime)
tp := broker.TopicPartition{
Namespace: namespace,
Topic: topic,
Partition: int32(i),
}
grpcClientConn, err := mc.findBroker(tp)
if err != nil {
return nil, err
}
client, err := setupSubscriberClient(grpcClientConn, tp, subscriberId, startTime)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -36,22 +45,7 @@ func (mc *MessagingClient) NewSubscriber(subscriberId, namespace, topic string,
}, nil }, nil
} }
func (mc *MessagingClient) setupSubscriberClient(subscriberId, namespace, topic string, partition int32, startTime time.Time) (messaging_pb.SeaweedMessaging_SubscribeClient, error) {
stream, err := setupSubscriberClient(subscriberId, namespace, topic, partition, startTime)
if err != nil {
return stream, err
}
if newBroker != nil {
}
return stream, nil
}
*/
func setupSubscriberClient(grpcConnection *grpc.ClientConn, subscriberId string, namespace string, topic string, partition int32, startTime time.Time) (stream messaging_pb.SeaweedMessaging_SubscribeClient, err error) {
func setupSubscriberClient(grpcConnection *grpc.ClientConn, tp broker.TopicPartition, subscriberId string, startTime time.Time) (stream messaging_pb.SeaweedMessaging_SubscribeClient, err error) {
stream, err = messaging_pb.NewSeaweedMessagingClient(grpcConnection).Subscribe(context.Background()) stream, err = messaging_pb.NewSeaweedMessagingClient(grpcConnection).Subscribe(context.Background())
if err != nil { if err != nil {
return return
@ -60,9 +54,9 @@ func setupSubscriberClient(grpcConnection *grpc.ClientConn, subscriberId string,
// send init message // send init message
err = stream.Send(&messaging_pb.SubscriberMessage{ err = stream.Send(&messaging_pb.SubscriberMessage{
Init: &messaging_pb.SubscriberMessage_InitMessage{ Init: &messaging_pb.SubscriberMessage_InitMessage{
Namespace: namespace,
Topic: topic,
Partition: partition,
Namespace: tp.Namespace,
Topic: tp.Topic,
Partition: tp.Partition,
StartPosition: messaging_pb.SubscriberMessage_InitMessage_TIMESTAMP, StartPosition: messaging_pb.SubscriberMessage_InitMessage_TIMESTAMP,
TimestampNs: startTime.UnixNano(), TimestampNs: startTime.UnixNano(),
SubscriberId: subscriberId, SubscriberId: subscriberId,
@ -90,6 +84,10 @@ func (s *Subscriber) doSubscribe(partition int, processFn func(m *messaging_pb.M
println(listenErr.Error()) println(listenErr.Error())
return listenErr return listenErr
} }
if resp.Data == nil {
// this could be heartbeat from broker
continue
}
processFn(resp.Data) processFn(resp.Data)
} }
} }

Loading…
Cancel
Save