util

Documentation

Index

• Constants
• Variables
• func EnvBoolVar(value *bool, key string)
• func EnvIntVar(value *int, key string)
• func EnvStringVar(value *string, key string)
• func GetIP4Byname(host string) (ips []string, err error)
• func GetNetAddrPort(addr net.Addr) (port int)
• func GetOutboundIP() (ip net.IP, err error)
• func GetShift(s int) (shift uint)
• func GetSourceName(name string) (sourcename string)
• func GetSpareTCPPort(portBegin int) (port int)
• func InitGlobalParsingPool()
• func InitGlobalTimerWheel()
• func InitGlobalWritingPool(maxWorkers int)
• func InitLogger(newLogPaths []string)
• func JksToPem(jksPath, jksPassword string, overwrite bool) (certPemPath, keyPemPath string, err error)
• func NewTLSConfig(caCertFiles, clientCertFile, clientKeyFile string, insecureSkipVerify bool) (*tls.Config, error)
• func Run(appName string, initFunc, jobFunc, cleanupFunc func() error)
• func SetLogLevel(newLogLevel string)
• func StringContains(arr []string, str string) bool
• func WaitForExitSign()
• type WorkerPool
  • func NewWorkerPool(maxWorkers int, queueSize int) *WorkerPool
  • func (w *WorkerPool) Resize(maxWorkers int)
  • func (w *WorkerPool) Restart()
  • func (w *WorkerPool) StopWait()
  • func (w *WorkerPool) Submit(fn func()) (err error)

Constants

const (
	StateRunning uint32 = 0
	StateStopped uint32 = 1
)

Variables

var (
	GlobalTimerWheel  *goetty.TimeoutWheel //the global timer wheel
	GlobalParsingPool *WorkerPool          //for all tasks' parsing, cpu intensive
	GlobalWritingPool *WorkerPool          //the all tasks' writing ClickHouse, cpu-net balance
	Logger            *zap.Logger
)

var (
	// ErrStopped when stopped
	ErrStopped = errors.New("WorkerPool already stopped")
)

Functions

func EnvBoolVar

func EnvBoolVar(value *bool, key string)

func EnvIntVar

func EnvIntVar(value *int, key string)

func EnvStringVar

func EnvStringVar(value *string, key string)

func GetIP4Byname

func GetIP4Byname(host string) (ips []string, err error)

func GetNetAddrPort

func GetNetAddrPort(addr net.Addr) (port int)

https://stackoverflow.com/questions/50428176/how-to-get-ip-and-port-from-net-addr-when-it-could-be-a-net-udpaddr-or-net-tcpad

func GetOutboundIP

func GetOutboundIP() (ip net.IP, err error)

GetOutboundIP get preferred outbound ip of this machine https://stackoverflow.com/questions/23558425/how-do-i-get-the-local-ip-address-in-go

func GetShift

func GetShift(s int) (shift uint)

GetShift returns the smallest `shift` which 1<<shift is no smaller than s

func GetSourceName

func GetSourceName(name string) (sourcename string)

GetSourceName returns the field name in message for the given ClickHouse column

func GetSpareTCPPort

func GetSpareTCPPort(portBegin int) (port int)

GetSpareTCPPort find a spare TCP port

func InitGlobalParsingPool

func InitGlobalParsingPool()

InitGlobalParsingPool initialize GlobalParsingPool

func InitGlobalTimerWheel

func InitGlobalTimerWheel()

InitGlobalTimerWheel initialize the global timer wheel

func InitGlobalWritingPool

func InitGlobalWritingPool(maxWorkers int)

InitGlobalWritingPool initialize GlobalWritingPool

func InitLogger

func InitLogger(newLogPaths []string)

func JksToPem

func JksToPem(jksPath, jksPassword string, overwrite bool) (certPemPath, keyPemPath string, err error)

JksToPem converts JKS to PEM Refers to: https://serverfault.com/questions/715827/how-to-generate-key-and-crt-file-from-jks-file-for-httpd-apache-server

func NewTLSConfig

func NewTLSConfig(caCertFiles, clientCertFile, clientKeyFile string, insecureSkipVerify bool) (*tls.Config, error)

Refers to: https://medium.com/processone/using-tls-authentication-for-your-go-kafka-client-3c5841f2a625 https://github.com/denji/golang-tls https://www.baeldung.com/java-keystore-truststore-difference

func Run

func Run(appName string, initFunc, jobFunc, cleanupFunc func() error)

func SetLogLevel

func SetLogLevel(newLogLevel string)

func StringContains

func StringContains(arr []string, str string) bool

StringContains check if contains string in array

func WaitForExitSign

func WaitForExitSign()

Types

type WorkerPool

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

WorkerPool is a blocked worker pool inspired by https://github.com/gammazero/workerpool/

func NewWorkerPool

func NewWorkerPool(maxWorkers int, queueSize int) *WorkerPool

New creates and starts a pool of worker goroutines.

func (*WorkerPool) Resize

func (w *WorkerPool) Resize(maxWorkers int)

Resize ensures worker number match the expected one.

func (*WorkerPool) Restart

func (w *WorkerPool) Restart()

func (*WorkerPool) StopWait

func (w *WorkerPool) StopWait()

StopWait stops the worker pool and waits for all queued tasks tasks to complete.

func (*WorkerPool) Submit

func (w *WorkerPool) Submit(fn func()) (err error)

Submit enqueues a function for a worker to execute. Submit will block regardless if there is no free workers.