refactor: type-specific 'handlers'

internal/background_flusher.go

@@ -13,11 +13,11 @@ type BackgroundFlusher interface {
 type backgroundFlusher struct {
 	ticker   *time.Ticker
 	interval time.Duration
-	handler  LineHandler
+	handler  BatchBuilder
 	stop     chan struct{}
 }
 
-func NewBackgroundFlusher(interval time.Duration, handler LineHandler) BackgroundFlusher {
+func NewBackgroundFlusher(interval time.Duration, handler BatchBuilder) BackgroundFlusher {
 	return &backgroundFlusher{
 		interval: interval,
 		handler:  handler,

internal/delta/formatter.go

@@ -0,0 +1,18 @@
+package delta
+
+import (
+	"fmt"
+
+	"github.com/wavefronthq/wavefront-sdk-go/internal"
+	"github.com/wavefronthq/wavefront-sdk-go/internal/metric"
+)
+
+func Line(name string, value float64, source string, tags map[string]string, defaultSource string) (string, error) {
+	if name == "" {
+		return "", fmt.Errorf("empty metric name")
+	}
+	if !internal.HasDeltaPrefix(name) {
+		name = internal.DeltaCounterName(name)
+	}
+	return metric.Line(name, value, 0, source, tags, defaultSource)
+}

internal/handler_factory.go

@@ -6,32 +6,54 @@
 	"github.com/wavefronthq/wavefront-sdk-go/internal/sdkmetrics"
 )
 
-type HandlerFactory struct {
+type SenderFactory struct {
 	metricsReporter    Reporter
 	tracesReporter     Reporter
 	flushInterval      time.Duration
 	bufferSize         int
-	lineHandlerOptions []LineHandlerOption
+	lineHandlerOptions []BatchAccumulatorOption
+	registry           sdkmetrics.Registry
 }
 
-func NewHandlerFactory(
+func NewSenderFactory(
 	metricsReporter,
 	tracesReporter Reporter,
 	flushInterval time.Duration,
 	bufferSize int,
-	registry sdkmetrics.Registry) *HandlerFactory {
-	return &HandlerFactory{
+	registry sdkmetrics.Registry) *SenderFactory {
+	return &SenderFactory{
+		registry:        registry,
 		metricsReporter: metricsReporter,
 		tracesReporter:  tracesReporter,
 		flushInterval:   flushInterval,
 		bufferSize:      bufferSize,
-		lineHandlerOptions: []LineHandlerOption{
+		lineHandlerOptions: []BatchAccumulatorOption{
 			SetRegistry(registry),
 		},
 	}
 }
 
-func (f *HandlerFactory) NewPointHandler(batchSize int) *RealLineHandler {
+func (f *SenderFactory) NewPointSender(batchSize int) TypedSender {
+	return NewTypedSender(f.registry.PointsTracker(), f.newPointHandler(batchSize))
+}
+
+func (f *SenderFactory) NewHistogramSender(batchSize int) TypedSender {
+	return NewTypedSender(f.registry.HistogramsTracker(), f.NewHistogramHandler(batchSize))
+}
+
+func (f *SenderFactory) NewSpanSender(batchSize int) TypedSender {
+	return NewTypedSender(f.registry.SpansTracker(), f.newSpanHandler(batchSize))
+}
+
+func (f *SenderFactory) NewEventsSender() TypedSender {
+	return NewTypedSender(f.registry.EventsTracker(), f.newEventHandler())
+}
+
+func (f *SenderFactory) NewSpanLogSender(batchSize int) TypedSender {
+	return NewTypedSender(f.registry.SpanLogsTracker(), f.newSpanLogHandler(batchSize))
+}
+
+func (f *SenderFactory) newPointHandler(batchSize int) *RealBatchBuilder {
 	return NewLineHandler(
 		f.metricsReporter,
 		metricFormat,
@@ -43,7 +65,7 @@
 	)
 }
 
-func (f *HandlerFactory) NewHistogramHandler(batchSize int) *RealLineHandler {
+func (f *SenderFactory) NewHistogramHandler(batchSize int) *RealBatchBuilder {
 	return NewLineHandler(
 		f.metricsReporter,
 		histogramFormat,
@@ -55,7 +77,7 @@
 	)
 }
 
-func (f *HandlerFactory) NewSpanHandler(batchSize int) *RealLineHandler {
+func (f *SenderFactory) newSpanHandler(batchSize int) *RealBatchBuilder {
 	return NewLineHandler(
 		f.tracesReporter,
 		traceFormat,
@@ -67,7 +89,7 @@
 	)
 }
 
-func (f *HandlerFactory) NewSpanLogHandler(batchSize int) *RealLineHandler {
+func (f *SenderFactory) newSpanLogHandler(batchSize int) *RealBatchBuilder {
 	return NewLineHandler(
 		f.tracesReporter,
 		spanLogsFormat,
@@ -79,10 +101,10 @@
 	)
 }
 
-// NewEventHandler creates a RealLineHandler for the Event type
+// NewEventHandler creates a RealBatchBuilder for the Event type
 // The Event handler always sets "ThrottleRequestsOnBackpressure" to true
 // And always uses a batch size of exactly 1.
-func (f *HandlerFactory) NewEventHandler() *RealLineHandler {
+func (f *SenderFactory) newEventHandler() *RealBatchBuilder {
 	return NewLineHandler(
 		f.metricsReporter,
 		eventFormat,

internal/interfaces.go

@@ -24,7 +24,7 @@ type ConnectionHandler interface {
 	Flusher
 }
 
-type LineHandler interface {
+type BatchBuilder interface {
 	HandleLine(line string) error
 	Start()
 	Stop()

internal/real_line_handler.go → internal/real_batch_builder.go

@@ -22,7 +22,7 @@
 	defaultThrottledSleepDuration = time.Second * 30
 )
 
-type RealLineHandler struct {
+type RealBatchBuilder struct {
 	// keep these two fields as first element of struct
 	// to guarantee 64-bit alignment on 32-bit machines.
 	// atomic.* functions crash if operands are not 64-bit aligned.
@@ -46,34 +46,34 @@
 	resumeAt time.Time
 }
 
 func (lh *RealLineHandler) Format() string {
 	return lh.format
 }
 
 var errThrottled = errors.New("error: throttled event creation")
 
-type LineHandlerOption func(*RealLineHandler)
+type BatchAccumulatorOption func(*RealBatchBuilder)
 
-func SetRegistry(registry sdkmetrics.Registry) LineHandlerOption {
-	return func(handler *RealLineHandler) {
+func SetRegistry(registry sdkmetrics.Registry) BatchAccumulatorOption {
+	return func(handler *RealBatchBuilder) {
 		handler.internalRegistry = registry
 	}
 }
 
-func SetHandlerPrefix(prefix string) LineHandlerOption {
-	return func(handler *RealLineHandler) {
+func SetHandlerPrefix(prefix string) BatchAccumulatorOption {
+	return func(handler *RealBatchBuilder) {
 		handler.prefix = prefix
 	}
 }
 
-func ThrottleRequestsOnBackpressure() LineHandlerOption {
-	return func(handler *RealLineHandler) {
+func ThrottleRequestsOnBackpressure() BatchAccumulatorOption {
+	return func(handler *RealBatchBuilder) {
 		handler.throttleOnBackpressure = true
 	}
 }
 
-func NewLineHandler(reporter Reporter, format string, flushInterval time.Duration, batchSize, maxBufferSize int, setters ...LineHandlerOption) *RealLineHandler {
-	lh := &RealLineHandler{
+func NewLineHandler(reporter Reporter, format string, flushInterval time.Duration, batchSize, maxBufferSize int, setters ...BatchAccumulatorOption) *RealBatchBuilder {
+	lh := &RealBatchBuilder{
 		Reporter:               reporter,
 		BatchSize:              batchSize,
 		MaxBufferSize:          maxBufferSize,
@@ -82,7 +82,7 @@
 	}
 
 	lh.buffer = make(chan string, lh.MaxBufferSize)
 	lh.flusher = NewBackgroundFlusher(flushInterval, lh)
 
 	for _, setter := range setters {
 		setter(lh)
@@ -99,11 +99,11 @@
 	return lh
 }
 
-func (lh *RealLineHandler) Start() {
+func (lh *RealBatchBuilder) Start() {
 	lh.flusher.Start()
 }
 
-func (lh *RealLineHandler) HandleLine(line string) error {
+func (lh *RealBatchBuilder) HandleLine(line string) error {
 	select {
 	case lh.buffer <- line:
 		return nil
@@ -120,7 +120,7 @@
 	return y
 }
 
-func (lh *RealLineHandler) flush() error {
+func (lh *RealBatchBuilder) flush() error {
 	lh.mtx.Lock()
 	defer lh.mtx.Unlock()
 	bufLen := len(lh.buffer)
@@ -135,7 +135,7 @@
 	return nil
 }
 
-func (lh *RealLineHandler) FlushWithThrottling() error {
+func (lh *RealBatchBuilder) FlushWithThrottling() error {
 	if time.Now().Before(lh.resumeAt) {
 		log.Println("attempting to flush, but flushing is currently throttled by the server")
 		log.Printf("sleeping until: %s\n", lh.resumeAt.Format(time.RFC3339))
@@ -144,7 +144,7 @@
 	return lh.Flush()
 }
 
-func (lh *RealLineHandler) Flush() error {
+func (lh *RealBatchBuilder) Flush() error {
 	flushErr := lh.flush()
 	if flushErr == errThrottled && lh.throttleOnBackpressure {
 		atomic.AddInt64(&lh.throttled, 1)
@@ -154,7 +154,7 @@
 	return flushErr
 }
 
-func (lh *RealLineHandler) FlushAll() error {
+func (lh *RealBatchBuilder) FlushAll() error {
 	lh.mtx.Lock()
 	defer lh.mtx.Unlock()
 	bufLen := len(lh.buffer)
@@ -178,7 +178,7 @@
 	return nil
 }
 
-func (lh *RealLineHandler) report(lines []string) error {
+func (lh *RealBatchBuilder) report(lines []string) error {
 	strLines := strings.Join(lines, "")
 	resp, err := lh.Reporter.Report(lh.format, strLines)
 
@@ -208,23 +208,23 @@
 	return true
 }
 
-func (lh *RealLineHandler) bufferLines(batch []string) {
+func (lh *RealBatchBuilder) bufferLines(batch []string) {
 	log.Println("error reporting to Wavefront. buffering lines.")
 	for _, line := range batch {
 		_ = lh.HandleLine(line)
 	}
 }
 
-func (lh *RealLineHandler) GetFailureCount() int64 {
+func (lh *RealBatchBuilder) GetFailureCount() int64 {
 	return atomic.LoadInt64(&lh.failures)
 }
 
 // GetThrottledCount returns the number of Throttled errors received.
-func (lh *RealLineHandler) GetThrottledCount() int64 {
+func (lh *RealBatchBuilder) GetThrottledCount() int64 {
 	return atomic.LoadInt64(&lh.throttled)
 }
 
-func (lh *RealLineHandler) Stop() {
+func (lh *RealBatchBuilder) Stop() {
 	lh.flusher.Stop()
 	if err := lh.FlushAll(); err != nil {
 		log.Println(err)

internal/real_line_handler_test.go → internal/real_batch_builder_test.go

@@ -94,7 +94,7 @@ func TestHandleLine_OnAuthError_DoNotBuffer(t *testing.T) {
 }
 
 func TestFlushWithThrottling_WhenThrottling_DelayUntilThrottleInterval(t *testing.T) {
-	lh := &RealLineHandler{
+	lh := &RealBatchBuilder{
 		Reporter:               &fakeReporter{},
 		MaxBufferSize:          100,
 		BatchSize:              10,
@@ -117,7 +117,7 @@ func TestFlushWithThrottling_WhenThrottling_DelayUntilThrottleInterval(t *testin
 }
 
 func TestBackgroundFlushWithThrottling_WhenThrottling_DelayUntilThrottleInterval(t *testing.T) {
-	lh := &RealLineHandler{
+	lh := &RealBatchBuilder{
 		Reporter:               &fakeReporter{},
 		MaxBufferSize:          100,
 		BatchSize:              10,
@@ -142,7 +142,7 @@ func TestBackgroundFlushWithThrottling_WhenThrottling_DelayUntilThrottleInterval
 func TestFlushTicker_WhenThrottlingEnabled_AndReceives406Error_ThrottlesRequestsUntilNextSleepDuration(t *testing.T) {
 	throttledSleepDuration := 250 * time.Millisecond
 	briskTickTime := 50 * time.Millisecond
-	lh := &RealLineHandler{
+	lh := &RealBatchBuilder{
 		Reporter:               &fakeReporter{},
 		MaxBufferSize:          100,
 		BatchSize:              10,
@@ -194,7 +194,7 @@ func checkLength(buffer chan string, length int, msg string, t *testing.T) {
 	}
 }
 
-func addLines(lh *RealLineHandler, linesToAdd int, expectedLen int, t *testing.T) {
+func addLines(lh *RealBatchBuilder, linesToAdd int, expectedLen int, t *testing.T) {
 	for i := 0; i < linesToAdd; i++ {
 		err := lh.HandleLine("dummyLine")
 		if err != nil {
@@ -214,8 +214,8 @@ func makeBuffer(num int) []string {
 	return buf
 }
 
-func makeLineHandler(bufSize, batchSize int) *RealLineHandler {
-	return &RealLineHandler{
+func makeLineHandler(bufSize, batchSize int) *RealBatchBuilder {
+	return &RealBatchBuilder{
 		Reporter:      &fakeReporter{},
 		MaxBufferSize: bufSize,
 		BatchSize:     batchSize,

internal/reporter.go

@@ -10,24 +10,24 @@ import (
 	"github.com/wavefronthq/wavefront-sdk-go/internal/auth"
 )
 
-// The implementation of a Reporter that reports points directly to a Wavefront server.
-type reporter struct {
+// batchingHTTPReporter is a Reporter that reports points in batches via HTTP(S)
+type batchingHTTPReporter struct {
 	serverURL    string
 	tokenService auth.Service
 	client       *http.Client
 }
 
 // NewReporter creates a metrics Reporter
 func NewReporter(server string, tokenService auth.Service, client *http.Client) Reporter {
-	return &reporter{
+	return &batchingHTTPReporter{
 		serverURL:    server,
 		tokenService: tokenService,
 		client:       client,
 	}
 }
 
 // Report creates and sends a POST to the reportEndpoint with the given pointLines
-func (reporter reporter) Report(format string, pointLines string) (*http.Response, error) {
+func (reporter batchingHTTPReporter) Report(format string, pointLines string) (*http.Response, error) {
 	if format == "" || pointLines == "" {
 		return nil, formatError
 	}
@@ -63,7 +63,7 @@ func linesToGzippedBytes(pointLines string) ([]byte, error) {
 	return buf.Bytes(), err
 }
 
-func (reporter reporter) buildRequest(format string, body []byte) (*http.Request, error) {
+func (reporter batchingHTTPReporter) buildRequest(format string, body []byte) (*http.Request, error) {
 	apiURL := reporter.serverURL + reportEndpoint
 	req, err := http.NewRequest("POST", apiURL, bytes.NewBuffer(body))
 	if err != nil {
@@ -84,7 +84,7 @@ func (reporter reporter) buildRequest(format string, body []byte) (*http.Request
 	return req, nil
 }
 
-func (reporter reporter) reportEvent(event string) (*http.Response, error) {
+func (reporter batchingHTTPReporter) reportEvent(event string) (*http.Response, error) {
 	if event == "" {
 		return nil, formatError
 	}
@@ -106,7 +106,7 @@ func (reporter reporter) reportEvent(event string) (*http.Response, error) {
 	return reporter.execute(req)
 }
 
-func (reporter reporter) execute(req *http.Request) (*http.Response, error) {
+func (reporter batchingHTTPReporter) execute(req *http.Request) (*http.Response, error) {
 	resp, err := reporter.client.Do(req)
 	if err != nil {
 		return nil, err
@@ -116,10 +116,10 @@ func (reporter reporter) execute(req *http.Request) (*http.Response, error) {
 	return resp, nil
 }
 
-func (reporter reporter) Close() {
+func (reporter batchingHTTPReporter) Close() {
 	reporter.tokenService.Close()
 }
 
-func (reporter reporter) IsDirect() bool {
+func (reporter batchingHTTPReporter) IsDirect() bool {
 	return reporter.tokenService.IsDirect()
 }