package broker

import "go.gazette.dev/core/broker"

Package broker implements the broker runtime and protocol.JournalServer APIs (Read, Append, Replicate, List, Apply). Its `pipeline` type manages the coordination of write transactions, and `resolver` the mapping of journal names to Routes of responsible brokers. `replica` is a top-level collection of runtime state and maintenance tasks associated with the processing of a journal. gRPC proxy support is also implemented by this package.

Index ¶

• Variables
• func JournalIsConsistent(item allocator.Item, _ keyspace.KeyValue, assignments keyspace.KeyValues) bool
• func JournalRouteMatchesAssignments(rt pb.Route, assignments keyspace.KeyValues) bool
• func NewKeySpace(prefix string) *keyspace.KeySpace
• func SetSharedPersister(p *fragment.Persister)
• type Service
• func NewService(state *allocator.State, jc pb.JournalClient, etcd *clientv3.Client) *Service
• func (svc *Service) Append(stream pb.Journal_AppendServer) (err error)
• func (svc *Service) Apply(ctx context.Context, req *pb.ApplyRequest) (resp *pb.ApplyResponse, err error)
• func (svc *Service) IsNoopRouter() bool
• func (svc *Service) List(ctx context.Context, req *pb.ListRequest) (resp *pb.ListResponse, err error)
• func (svc *Service) ListFragments(ctx context.Context, req *pb.FragmentsRequest) (resp *pb.FragmentsResponse, err error)
• func (svc *Service) QueueTasks(tasks *task.Group, server *server.Server, finishFn func())
• func (svc *Service) Read(req *pb.ReadRequest, stream pb.Journal_ReadServer) (err error)
• func (svc *Service) Replicate(stream pb.Journal_ReplicateServer) (err error)
• func (svc *Service) Route(ctx context.Context, item string) pb.Route
• func (svc *Service) UpdateRoute(string, *pb.Route)

Package Files ¶

append_api.go append_flow_control.go append_fsm.go doc.go fragments_api.go key_space.go list_apply_api.go pipeline.go read_api.go replica.go replicate_api.go resolver.go service.go

Variables ¶

var (
    // MaxAppendRate is the maximum rate at which any journal may be appended to,
    // in bytes per second. It upper-bounds the MaxAppendRate of any particular
    // JournalSpec. If zero, there is no maximum rate.
    MaxAppendRate int64 = 0 // No limit.
    // MinAppendRate is the minimum rate at which any Append RPC client may
    // stream content chunks, in bytes per second. Client RPCs that are unable
    // to sustain this flow rate in any given second are aborted, allowing other
    // blocked RPCs to proceed. MinAppendRate provides baseline protection to
    // limit the impact of slow or faulted clients over the pipeline, which is
    // an exclusively owned and highly contended resource.
    MinAppendRate int64 = 1 << 16 // 64K per second.
    // ErrFlowControlUnderflow is returned if an Append RPC was terminated due to
    // flow control policing. Specifically, the client failed to sustain the
    // MinAppendRate when sending content chunks of the stream.
    ErrFlowControlUnderflow = errors.New(
        "client stream didn't sustain the minimum append data rate")
)

func JournalIsConsistent ¶ Uses

func JournalIsConsistent(item allocator.Item, _ keyspace.KeyValue, assignments keyspace.KeyValues) bool

JournalIsConsistent returns true if all allocator.Assignments of the JournalSpec identified by Item advertise the same Route, denoting that all replicas of the journal have synchronized.

func JournalRouteMatchesAssignments ¶ Uses

func JournalRouteMatchesAssignments(rt pb.Route, assignments keyspace.KeyValues) bool

JournalRouteMatchesAssignments returns true iff the Route is equivalent to the Route marshaled with each of the journal's |assignments|.

func NewKeySpace ¶ Uses

func NewKeySpace(prefix string) *keyspace.KeySpace

NewKeySpace returns a KeySpace suitable for use with an Allocator. It decodes allocator Items as JournalSpec messages, Members as BrokerSpecs, and Assignments as Routes.

func SetSharedPersister ¶ Uses

func SetSharedPersister(p *fragment.Persister)

SetSharedPersister sets the Persister instance used by the `broker` package.

type Service ¶ Uses

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

Service is the top-level runtime concern of a Gazette Broker process. It drives local journal handling in response to allocator.State, powers journal resolution, and is also an implementation of protocol.JournalServer.

func NewService ¶ Uses

func NewService(state *allocator.State, jc pb.JournalClient, etcd *clientv3.Client) *Service

NewService constructs a new broker Service, driven by allocator.State.

func (*Service) Append ¶ Uses

func (svc *Service) Append(stream pb.Journal_AppendServer) (err error)

Append dispatches the JournalServer.Append API.

func (*Service) Apply ¶ Uses

func (svc *Service) Apply(ctx context.Context, req *pb.ApplyRequest) (resp *pb.ApplyResponse, err error)

Apply dispatches the JournalServer.Apply API.

func (*Service) IsNoopRouter ¶ Uses

func (svc *Service) IsNoopRouter() bool

IsNoopRouter returns false.

func (*Service) List ¶ Uses

func (svc *Service) List(ctx context.Context, req *pb.ListRequest) (resp *pb.ListResponse, err error)

List dispatches the JournalServer.List API.

func (*Service) ListFragments ¶ Uses

func (svc *Service) ListFragments(ctx context.Context, req *pb.FragmentsRequest) (resp *pb.FragmentsResponse, err error)

ListFragments dispatches the JournalServer.ListFragments API.

func (*Service) QueueTasks ¶ Uses

func (svc *Service) QueueTasks(tasks *task.Group, server *server.Server, finishFn func())

QueueTasks of the Service to watch its KeySpace and serve local replicas.

func (*Service) Read ¶ Uses

func (svc *Service) Read(req *pb.ReadRequest, stream pb.Journal_ReadServer) (err error)

Read dispatches the JournalServer.Read API.

func (*Service) Replicate ¶ Uses

func (svc *Service) Replicate(stream pb.Journal_ReplicateServer) (err error)

Replicate dispatches the JournalServer.Replicate API.

func (*Service) Route ¶ Uses

func (svc *Service) Route(ctx context.Context, item string) pb.Route

Route an item using the Service resolver. Route implements the protocol.DispatchRouter interface, and enables usages of protocol.WithDispatchItemRoute (eg, `client` & `http_gateway` packages) to resolve items via the Service resolver.

func (*Service) UpdateRoute ¶ Uses

func (svc *Service) UpdateRoute(string, *pb.Route)

UpdateRoute is a no-op implementation of protocol.DispatchRouter.

Directories ¶