pulsardriver

Documentation

Index

• Variables
• func CancelPulsarConsumer(key string)
• func GetPulsarClient(pulsarURL, pulsarToken string, reset bool) (pulsar.Client, error)
• func GetPulsarConsumer(pulsarURL, pulsarToken, topic, subName, subInitPos, subType, subKey string) (pulsar.Consumer, error)
• func GetPulsarProducer(pulsarURL, pulsarToken, topic string) (pulsar.Producer, error)
• func NewPulsarClient(url, tokenStr string) (pulsar.Client, error)
• func SendToPulsar(url, token, topic string, data []byte, async bool) error
• type PulsarClient
  • func (c *PulsarClient) Close()
  • func (c *PulsarClient) GetClient(url, tokenStr string) (pulsar.Client, error)
  • func (c *PulsarClient) Reconnect() (pulsar.Client, error)
  • func (c *PulsarClient) UpdateTime()
• type PulsarConsumer
  • func (c *PulsarConsumer) Close()
  • func (c *PulsarConsumer) GetConsumer() (pulsar.Consumer, error)
  • func (c *PulsarConsumer) Reconnect() (pulsar.Consumer, error)
  • func (c *PulsarConsumer) UpdateTime()
• type PulsarProducer
  • func (c *PulsarProducer) Close()
  • func (c *PulsarProducer) GetProducer() (pulsar.Producer, error)
  • func (c *PulsarProducer) Reconnect() (pulsar.Producer, error)
  • func (c *PulsarProducer) UpdateTime()

Variables

var ClientCache = make(map[string]*PulsarClient)

ClientCache caches a list Pulsar clients

var ConsumerCache = make(map[string]*PulsarConsumer)

ConsumerCache caches a list Pulsar prudcers key is a string concatenated with pulsar url, token, and topic full name

var ProducerCache = util.NewCache(util.CacheOption{
	TTL:           time.Duration(producerCacheTTL) * time.Second,
	CleanInterval: time.Duration(producerCacheTTL+2) * time.Second,
	ExpireCallback: func(key string, value interface{}) {
		if obj, ok := value.(*PulsarProducer); ok {
			obj.Close()
		} else {
			log.Errorf("wrong PulsarProducer object type stored in Cache")
		}
	},
})

ProducerCache is the cache for Producer objects

Functions

func CancelPulsarConsumer

func CancelPulsarConsumer(key string)

CancelPulsarConsumer closes Pulsar consumer and removes from the ConsumerCache

func GetPulsarClient

func GetPulsarClient(pulsarURL, pulsarToken string, reset bool) (pulsar.Client, error)

GetPulsarClient gets a Pulsar client object

func GetPulsarConsumer

func GetPulsarConsumer(pulsarURL, pulsarToken, topic, subName, subInitPos, subType, subKey string) (pulsar.Consumer, error)

GetPulsarConsumer gets a Pulsar consumer object

func GetPulsarProducer

func GetPulsarProducer(pulsarURL, pulsarToken, topic string) (pulsar.Producer, error)

GetPulsarProducer gets a Pulsar producer object

func NewPulsarClient

func NewPulsarClient(url, tokenStr string) (pulsar.Client, error)

NewPulsarClient always creates a new pulsar.Client connection

func SendToPulsar

func SendToPulsar(url, token, topic string, data []byte, async bool) error

SendToPulsar sends data to a Pulsar producer.

Types

type PulsarClient

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

PulsarClient encapsulates the Pulsar Client object

func (*PulsarClient) Close

func (c *PulsarClient) Close()

Close closes the Pulsar client

func (*PulsarClient) GetClient

func (c *PulsarClient) GetClient(url, tokenStr string) (pulsar.Client, error)

GetClient acquires a new pulsar client

func (*PulsarClient) Reconnect

func (c *PulsarClient) Reconnect() (pulsar.Client, error)

Reconnect closes the current connection and reconnects again

func (*PulsarClient) UpdateTime

func (c *PulsarClient) UpdateTime()

UpdateTime updates all time stamps in the object

type PulsarConsumer

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

PulsarConsumer encapsulates the Pulsar Consumer object

func (*PulsarConsumer) Close

func (c *PulsarConsumer) Close()

Close closes the Pulsar client

func (*PulsarConsumer) GetConsumer

func (c *PulsarConsumer) GetConsumer() (pulsar.Consumer, error)

GetConsumer acquires a new pulsar consumer

func (*PulsarConsumer) Reconnect

func (c *PulsarConsumer) Reconnect() (pulsar.Consumer, error)

Reconnect closes the current connection and reconnects again

func (*PulsarConsumer) UpdateTime

func (c *PulsarConsumer) UpdateTime()

UpdateTime updates all time stamps in the object

type PulsarProducer

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

PulsarProducer encapsulates the Pulsar Producer object

func (*PulsarProducer) Close

func (c *PulsarProducer) Close()

Close closes the Pulsar client

func (*PulsarProducer) GetProducer

func (c *PulsarProducer) GetProducer() (pulsar.Producer, error)

GetProducer acquires a new pulsar producer

func (*PulsarProducer) Reconnect

func (c *PulsarProducer) Reconnect() (pulsar.Producer, error)

Reconnect closes the current connection and reconnects again

func (*PulsarProducer) UpdateTime

func (c *PulsarProducer) UpdateTime()

UpdateTime updates all time stamps in the object