pulsar

Documentation

Index

• type Athenz
• type Authentication
  • func (authentication *Authentication) Auth() pulsar.Authentication
• type FlusherPulsar
  • func (f *FlusherPulsar) Description() string
  • func (f *FlusherPulsar) Flush(projectName string, logstoreName string, configName string, ...) error
  • func (f *FlusherPulsar) Init(context pipeline.Context) error
  • func (f *FlusherPulsar) IsReady(projectName string, logstoreName string, logstoreKey int64) bool
  • func (*FlusherPulsar) SetUrgent(flag bool)
  • func (f *FlusherPulsar) Stop() error
  • func (f *FlusherPulsar) Validate() error
• type OAuth2
• type Producers
  • func NewProducers(context context.Context, maxProducers int) *Producers
  • func (p *Producers) Close() error
  • func (p *Producers) GetProducer(topic string, client pulsar.Client, producerOptions pulsar.ProducerOptions) (pulsar.Producer, error)
• type TLS
• type Token

Types

type Athenz

type Athenz struct {
	ProviderDomain  string
	TenantDomain    string
	TenantService   string
	PrivateKey      string
	KeyID           string
	PrincipalHeader string
	ZtsURL          string
}

type Authentication

type Authentication struct {
	TLS    *TLS
	Token  *Token
	Athenz *Athenz
	OAuth2 *OAuth2
}

func (*Authentication) Auth

func (authentication *Authentication) Auth() pulsar.Authentication

type FlusherPulsar

type FlusherPulsar struct {
	// URL for the Pulsar service
	URL string
	// SendTimeout send timeout
	SendTimeout time.Duration
	// OperationTimeout sets producer-create, subscribe and unsubscribe operations timeout (default: 30 seconds)
	OperationTimeout time.Duration
	// ConnectionTimeout timeout for the establishment of a TCP connection (default: 5 seconds)
	ConnectionTimeout time.Duration
	// MaxConnectionsPerBroker max number of connections to a single broker that will kept in the pool. (default: 1 connection)
	MaxConnectionsPerBroker int
	// Topic The name of the pulsar topic
	Topic string
	// Name  The producer name
	Name string
	// Convert  ilogtail data convert config
	Convert convertConfig
	// Configure whether the Pulsar client accept untrusted TLS certificate from broker (default: false)
	// EnableTLS
	EnableTLS bool
	// Set the path to the trusted TLS certificate file
	TLSTrustCertsFilePath string
	// Authentication support tls
	Authentication Authentication

	// MaxReconnectToBroker specifies the maximum retry number of reconnectToBroker. (default: ultimate)
	MaxReconnectToBroker *uint
	// DisableBlockIfQueueFull controls whether Send and SendAsync block if producer's message queue is full
	DisableBlockIfQueueFull bool
	// CompressionType  Codec used to produce messages,NONE,LZ4,ZLIB,ZSTD
	CompressionType string
	// HashingScheme is used to define the partition on where to publish a particular message
	HashingScheme string
	// BatchingMaxPublishDelay the batch push delay
	BatchingMaxPublishDelay time.Duration
	// BatchingMaxMessages maximum number of messages in a batch
	BatchingMaxMessages uint
	// MaxCacheProducers Specify the max cache(lru) producers of dynamic topic
	MaxCacheProducers int

	PartitionKeys []string
	ClientID      string
	// contains filtered or unexported fields
}

func (*FlusherPulsar) Description

func (f *FlusherPulsar) Description() string

func (*FlusherPulsar) Flush

func (f *FlusherPulsar) Flush(projectName string, logstoreName string, configName string, logGroupList []*protocol.LogGroup) error

func (*FlusherPulsar) Init

func (f *FlusherPulsar) Init(context pipeline.Context) error

func (*FlusherPulsar) IsReady

func (f *FlusherPulsar) IsReady(projectName string, logstoreName string, logstoreKey int64) bool

IsReady is ready to flush

func (*FlusherPulsar) SetUrgent

func (*FlusherPulsar) SetUrgent(flag bool)

func (*FlusherPulsar) Stop

func (f *FlusherPulsar) Stop() error

func (*FlusherPulsar) Validate

func (f *FlusherPulsar) Validate() error

type OAuth2

type OAuth2 struct {
	Enabled    bool
	IssuerURL  string
	Audience   string
	PrivateKey string
	Scope      string
}

type Producers

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

func NewProducers

func NewProducers(context context.Context, maxProducers int) *Producers

func (*Producers) Close

func (p *Producers) Close() error

func (*Producers) GetProducer

func (p *Producers) GetProducer(topic string, client pulsar.Client, producerOptions pulsar.ProducerOptions) (pulsar.Producer, error)

type TLS

type TLS struct {
	// Path to the TLS cert to use for TLS required connections. (optional)
	CertFile string
	// Path to the TLS key to use for TLS required connections. (optional)
	KeyFile string
}

TLS is the interface used to configure a tcp client or server from a `Config`

type Token

type Token struct {
	Token string
}