flow

package

Go to main page

Versions in this module

v0

v0.13.7

Jan 7, 2021

v0.13.6

Oct 2, 2020

v0.13.5

Sep 28, 2020

v0.13.4

Apr 22, 2020

Changes in this version

type ELasticTestHandler

+ CustomHandler func(w http.ResponseWriter, r *http.Request)

v0.13.3

Apr 14, 2020

Changes in this version

+ const DEFAULT_ELASTIC_DOCUMENT_TYPE

v0.13.2

Nov 7, 2019

Changes in this version

type ELasticTestHandler

+ ResponseBody []byte

v0.13.1

Oct 15, 2019

v0.13.0

Oct 9, 2019

Changes in this version

+ const ErrInitGrpc

+ const ErrRegisterGrpc

+ const ErrReverseProxy

+ const ProtoParseError

+ func ConvertTimberToEsDocumentString(timber pb.Timber, m *jsonpb.Marshaler) string

+ type ProducerService interface

+ Close func()

+ LaunchREST func() error

+ Start func() error

+ func NewProducerService(params map[string]interface{}) ProducerService

v0.12.2

Sep 2, 2019

v0.12.1

Jul 15, 2019

v0.12.0

Jul 15, 2019

Changes in this version

+ const ErrConsumerWorker

+ const ErrConvertKafkaMessage

+ const ErrElasticsearchClient

+ const ErrHaltWorker

+ const ErrKafkaRetryLimitReached

+ const ErrMakeKafkaAdmin

+ const ErrMakeNewTopicWorker

+ const ErrMakeSyncProducer

+ const ErrSpawnWorker

+ const ErrSpawnWorkerOnNewTopic

+ const ErrStore

+ const InvalidContextError

+ const JsonParseError

+ const MissingContextError

+ const PrefixEventGroupID

+ const RetrieveMessageFailedError

+ func Contains(s []string, e string) bool

+ func ConvertTimberToElasticDocument(timber Timber) map[string]interface

+ func ConvertTimberToKafkaMessage(timber Timber, topic string) *sarama.ProducerMessage

+ func GetApplicationSecretCollection() []string

+ func InstruApplicationSecret(appSecret string)

+ func NewDummyKafkaFactory() *dummyKafkaFactory

+ func NewDummyRateLimiter() *dummyRateLimiter

+ func NewElastic(retrierFunc *ElasticRetrier, esConfig esConfig, urls ...string) (client elasticClient, err error)

+ func NewEsConfig(indexMethod string, bulkSize int, flushMs time.Duration, printTPS bool) esConfig

+ type ApplicationSecretCollection struct

+ type BaritoConsumerService interface

+ Close func()

+ NewTopicEventWorker func() ConsumerWorker

+ Start func() error

+ WorkerMap func() map[string]ConsumerWorker

+ func NewBaritoConsumerService(factory KafkaFactory, groupID string, elasticURL string, topicSuffix string, ...) BaritoConsumerService

+ type BaritoProducerService interface

+ Close func()

+ ServeHTTP func(rw http.ResponseWriter, req *http.Request)

+ Start func() error

+ func NewBaritoProducerService(factory KafkaFactory, addr string, maxTps int, rateLimitResetInterval int, ...) BaritoProducerService

+ type ClusterConsumer interface

+ Close func() error

+ Errors func() <-chan error

+ MarkOffset func(msg *sarama.ConsumerMessage, metadata string)

+ Messages func() <-chan *sarama.ConsumerMessage

+ Notifications func() <-chan *cluster.Notification

+ type ConsumerWorker interface

+ Halt func()

+ IsStart func() bool

+ OnError func(f func(error))

+ OnNotification func(f func(*cluster.Notification))

+ OnSuccess func(f func(*sarama.ConsumerMessage))

+ Start func()

+ Stop func()

+ func NewConsumerWorker(name string, consumer ClusterConsumer) ConsumerWorker

+ type ELasticTestHandler struct

+ CreateAPIStatus int

+ ExistAPIStatus int

+ PostAPIStatus int

+ func (handler *ELasticTestHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)

+ type Elastic interface

+ NewClient func()

+ OnFailure func(f func(*Timber))

+ Store func(ctx context.Context, timber Timber)

+ type ElasticRetrier struct

+ func NewElasticRetrier(t time.Duration, f func(err error)) *ElasticRetrier

+ func (r *ElasticRetrier) Retry(ctx context.Context, retry int, req *http.Request, resp *http.Response, ...) (time.Duration, bool, error)

+ type KafkaAdmin interface

+ AddTopic func(topic string)

+ Close func()

+ CreateTopic func(topic string, numPartitions int32, replicationFactor int16) error

+ Exist func(topic string) bool

+ RefreshTopics func() error

+ SetTopics func([]string)

+ Topics func() []string

+ func NewKafkaAdmin(client sarama.Client) (admin KafkaAdmin, err error)

+ type KafkaFactory interface

+ MakeClusterConsumer func(groupID, topic string, initialOffset int64) (worker ClusterConsumer, err error)

+ MakeKafkaAdmin func() (admin KafkaAdmin, err error)

+ MakeSyncProducer func() (producer sarama.SyncProducer, err error)

+ func NewKafkaFactory(brokers []string, config *sarama.Config) KafkaFactory

+ type LeakyBucket struct

+ func NewLeakyBucket(max int) *LeakyBucket

+ func (b *LeakyBucket) IsFull() bool

+ func (b *LeakyBucket) Max() int

+ func (b *LeakyBucket) Token() int

+ func (b *LeakyBucket) UpdateMax(newMax int)

+ func (l *LeakyBucket) Refill()

+ func (l *LeakyBucket) Take(count int) bool

+ type ProduceResult struct

+ Topic string

+ type RateLimiter interface

+ Bucket func(topic string) *LeakyBucket

+ IsHitLimit func(topic string, count int, maxTokenIfNotExist int) bool

+ IsStart func() bool

+ PutBucket func(topic string, bucket *LeakyBucket)

+ Start func()

+ Stop func()

+ func NewRateLimiter(duration int) RateLimiter

+ type Timber map[string]interface

+ func (t Timber) Context() (ctx *TimberContext)

+ func (t Timber) InitContext() (err error)

+ func (t Timber) SetContext(ctx *TimberContext)

+ func (t Timber) SetTimestamp(timestamp string)

+ func (t Timber) Timestamp() (s string)

+ func ConvertBytesToTimber(data []byte) (timber Timber, err error)

+ func ConvertKafkaMessageToTimber(message *sarama.ConsumerMessage) (timber Timber, err error)

+ func ConvertRequestToTimber(req *http.Request) (Timber, error)

+ func NewTimber() Timber

+ type TimberCollection struct

+ Context TimberContext

+ Items []Timber

+ func ConvertBatchRequestToTimberCollection(req *http.Request) (TimberCollection, error)

+ func ConvertBytesToTimberCollection(data []byte) (timberCollection TimberCollection, err error)

+ type TimberContext struct

+ AppMaxTPS int

+ AppSecret string

+ ESDocumentType string

+ ESIndexPrefix string

+ KafkaPartition int32

+ KafkaReplicationFactor int16

+ KafkaTopic string

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL