network

Documentation

Index

• type Config
• type Server
  • func Run(config *Config, kfkProtocolConfig *codec.KafkaProtocolConfig, ...) (*Server, error)
  • func (s *Server) ApiVersions(frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) AuthFailed() ([]byte, gnet.Action)
  • func (s *Server) Authed(context *context.NetworkContext) bool
  • func (s *Server) Fetch(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) FindCoordinator(frame []byte, version int16, config *codec.KafkaProtocolConfig) ([]byte, gnet.Action)
  • func (s *Server) FindCoordinatorVersion(frame []byte, version int16, config *codec.KafkaProtocolConfig) ([]byte, gnet.Action)
  • func (s *Server) Heartbeat(frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) JoinGroup(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) LeaveGroup(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) ListOffsets(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) ListOffsetsVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) Metadata(frame []byte, version int16, config *codec.KafkaProtocolConfig) ([]byte, gnet.Action)
  • func (s *Server) OffsetCommit(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) OffsetCommitVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) OffsetFetch(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) OffsetFetchVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) OnClosed(c gnet.Conn, err error) (action gnet.Action)
  • func (s *Server) OnInitComplete(server gnet.Server) (action gnet.Action)
  • func (s *Server) OnOpened(c gnet.Conn) (out []byte, action gnet.Action)
  • func (s *Server) React(frame []byte, c gnet.Conn) ([]byte, gnet.Action)
  • func (s *Server) ReactApiVersion(frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) ReactFetchVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) ReactHeartbeatVersion(frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) ReactJoinGroupVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) ReactLeaveGroupVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) ReactMetadataVersion(frame []byte, version int16, config *codec.KafkaProtocolConfig) ([]byte, gnet.Action)
  • func (s *Server) ReactSaslHandshakeAuthVersion(frame []byte, version int16, context *context.NetworkContext) ([]byte, gnet.Action)
  • func (s *Server) ReactSaslVersion(frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) ReactSyncGroupVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) SaslAuthenticate(frame []byte, version int16, context *context.NetworkContext) ([]byte, gnet.Action)
  • func (s *Server) SaslHandshake(frame []byte, version int16) ([]byte, gnet.Action)
  • func (s *Server) SyncGroup(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

Types

type Config

type Config struct {
	ListenAddr string
	MultiCore  bool
}

type Server

type Server struct {
	*gnet.EventServer
	ConnMap sync.Map
	SaslMap sync.Map
	// contains filtered or unexported fields
}

func Run

func Run(config *Config, kfkProtocolConfig *codec.KafkaProtocolConfig, impl service.KfkServer) (*Server, error)

func (*Server) ApiVersions

func (s *Server) ApiVersions(frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) AuthFailed

func (s *Server) AuthFailed() ([]byte, gnet.Action)

func (*Server) Authed

func (s *Server) Authed(context *context.NetworkContext) bool

func (*Server) Fetch

func (s *Server) Fetch(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) FindCoordinator

func (s *Server) FindCoordinator(frame []byte, version int16, config *codec.KafkaProtocolConfig) ([]byte, gnet.Action)

func (*Server) FindCoordinatorVersion

func (s *Server) FindCoordinatorVersion(frame []byte, version int16, config *codec.KafkaProtocolConfig) ([]byte, gnet.Action)

func (*Server) Heartbeat

func (s *Server) Heartbeat(frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) JoinGroup

func (s *Server) JoinGroup(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) LeaveGroup

func (s *Server) LeaveGroup(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) ListOffsets

func (s *Server) ListOffsets(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) ListOffsetsVersion

func (s *Server) ListOffsetsVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) Metadata

func (s *Server) Metadata(frame []byte, version int16, config *codec.KafkaProtocolConfig) ([]byte, gnet.Action)

func (*Server) OffsetCommit

func (s *Server) OffsetCommit(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) OffsetCommitVersion

func (s *Server) OffsetCommitVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) OffsetFetch

func (s *Server) OffsetFetch(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) OffsetFetchVersion

func (s *Server) OffsetFetchVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) OnClosed

func (s *Server) OnClosed(c gnet.Conn, err error) (action gnet.Action)

func (*Server) OnInitComplete

func (s *Server) OnInitComplete(server gnet.Server) (action gnet.Action)

func (*Server) OnOpened

func (s *Server) OnOpened(c gnet.Conn) (out []byte, action gnet.Action)

func (*Server) React

func (s *Server) React(frame []byte, c gnet.Conn) ([]byte, gnet.Action)

React Kafka 协议格式为APIKey和API Version APIKey 样例: 00 12

func (*Server) ReactApiVersion

func (s *Server) ReactApiVersion(frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) ReactFetchVersion

func (s *Server) ReactFetchVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) ReactHeartbeatVersion

func (s *Server) ReactHeartbeatVersion(frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) ReactJoinGroupVersion

func (s *Server) ReactJoinGroupVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) ReactLeaveGroupVersion

func (s *Server) ReactLeaveGroupVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) ReactMetadataVersion

func (s *Server) ReactMetadataVersion(frame []byte, version int16, config *codec.KafkaProtocolConfig) ([]byte, gnet.Action)

func (*Server) ReactSaslHandshakeAuthVersion

func (s *Server) ReactSaslHandshakeAuthVersion(frame []byte, version int16, context *context.NetworkContext) ([]byte, gnet.Action)

func (*Server) ReactSaslVersion

func (s *Server) ReactSaslVersion(frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) ReactSyncGroupVersion

func (s *Server) ReactSyncGroupVersion(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) SaslAuthenticate

func (s *Server) SaslAuthenticate(frame []byte, version int16, context *context.NetworkContext) ([]byte, gnet.Action)

func (*Server) SaslHandshake

func (s *Server) SaslHandshake(frame []byte, version int16) ([]byte, gnet.Action)

func (*Server) SyncGroup

func (s *Server) SyncGroup(ctx *context.NetworkContext, frame []byte, version int16) ([]byte, gnet.Action)