broker

Documentation

Index

• Constants
• func PickMember(members []string, key []byte) string
• type Member
  • func (m Member) String() string
• type MessageBroker
  • func NewMessageBroker(option *MessageBrokerOption, grpcDialOption grpc.DialOption) (messageBroker *MessageBroker, err error)
  • func (broker *MessageBroker) AdjustedUrl(location *filer_pb.Location) string
  • func (broker *MessageBroker) ConfigureTopic(c context.Context, request *messaging_pb.ConfigureTopicRequest) (*messaging_pb.ConfigureTopicResponse, error)
  • func (broker *MessageBroker) DeleteTopic(c context.Context, request *messaging_pb.DeleteTopicRequest) (*messaging_pb.DeleteTopicResponse, error)
  • func (broker *MessageBroker) FindBroker(c context.Context, request *messaging_pb.FindBrokerRequest) (*messaging_pb.FindBrokerResponse, error)
  • func (broker *MessageBroker) GetTopicConfiguration(c context.Context, request *messaging_pb.GetTopicConfigurationRequest) (*messaging_pb.GetTopicConfigurationResponse, error)
  • func (broker *MessageBroker) Publish(stream messaging_pb.SeaweedMessaging_PublishServer) error
  • func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_SubscribeServer) error
  • func (broker *MessageBroker) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) (err error)
• type MessageBrokerOption
• type TopicControl
• type TopicManager
  • func NewTopicManager(messageBroker *MessageBroker) *TopicManager
  • func (tm *TopicManager) ListTopicPartitions() (tps []TopicPartition)
  • func (tm *TopicManager) ReleaseLock(partition TopicPartition, isPublisher bool)
  • func (tm *TopicManager) RequestLock(partition TopicPartition, topicConfig *messaging_pb.TopicConfiguration, ...) *TopicControl
• type TopicPartition
  • func (tp *TopicPartition) String() string

Constants

const (
	TopicPartitionFmt = "%s/%s_%02d"
)

Functions

func PickMember

func PickMember(members []string, key []byte) string

Types

type Member

type Member string

func (Member) String

func (m Member) String() string

type MessageBroker

type MessageBroker struct {
	// contains filtered or unexported fields
}

func NewMessageBroker

func NewMessageBroker(option *MessageBrokerOption, grpcDialOption grpc.DialOption) (messageBroker *MessageBroker, err error)

func (*MessageBroker) AdjustedUrl

func (broker *MessageBroker) AdjustedUrl(location *filer_pb.Location) string

func (*MessageBroker) ConfigureTopic

func (broker *MessageBroker) ConfigureTopic(c context.Context, request *messaging_pb.ConfigureTopicRequest) (*messaging_pb.ConfigureTopicResponse, error)

func (*MessageBroker) DeleteTopic

func (broker *MessageBroker) DeleteTopic(c context.Context, request *messaging_pb.DeleteTopicRequest) (*messaging_pb.DeleteTopicResponse, error)

func (*MessageBroker) FindBroker

func (broker *MessageBroker) FindBroker(c context.Context, request *messaging_pb.FindBrokerRequest) (*messaging_pb.FindBrokerResponse, error)

func (*MessageBroker) GetTopicConfiguration

func (broker *MessageBroker) GetTopicConfiguration(c context.Context, request *messaging_pb.GetTopicConfigurationRequest) (*messaging_pb.GetTopicConfigurationResponse, error)

func (*MessageBroker) Publish

func (broker *MessageBroker) Publish(stream messaging_pb.SeaweedMessaging_PublishServer) error

func (*MessageBroker) Subscribe

func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_SubscribeServer) error

func (*MessageBroker) WithFilerClient

func (broker *MessageBroker) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) (err error)

type MessageBrokerOption

type MessageBrokerOption struct {
	Filers             []string
	DefaultReplication string
	MaxMB              int
	Ip                 string
	Port               int
	Cipher             bool
}

type TopicControl

type TopicControl struct {
	sync.Mutex
	// contains filtered or unexported fields
}

type TopicManager

type TopicManager struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewTopicManager

func NewTopicManager(messageBroker *MessageBroker) *TopicManager

func (*TopicManager) ListTopicPartitions

func (tm *TopicManager) ListTopicPartitions() (tps []TopicPartition)

func (*TopicManager) ReleaseLock

func (tm *TopicManager) ReleaseLock(partition TopicPartition, isPublisher bool)

func (*TopicManager) RequestLock

func (tm *TopicManager) RequestLock(partition TopicPartition, topicConfig *messaging_pb.TopicConfiguration, isPublisher bool) *TopicControl

type TopicPartition

type TopicPartition struct {
	Namespace string
	Topic     string
	Partition int32
}

func (*TopicPartition) String

func (tp *TopicPartition) String() string