add new 'WithErrors' constructors

lib.go

@@ -104,6 +104,7 @@ func NewSenderWithNewDeviceWithErrors(dev *api.DeviceCreate, cfg *Config) (*Send
 	return cfg.startWithInternalErrors(client, d)
 }
 
+// NewSenderWithNewSiteAndDevice creates a new device and site then returns a flow Sender for that newly created device
 func NewSenderWithNewSiteAndDevice(siteAndDevice *api.SiteAndDeviceCreate, errors chan<- error, cfg *Config) (*Sender, error) {
 	client := cfg.client()
 	d, err := client.CreateDeviceAndSite(siteAndDevice)
@@ -114,11 +115,31 @@ func NewSenderWithNewSiteAndDevice(siteAndDevice *api.SiteAndDeviceCreate, error
 	return cfg.start(client, d, errors)
 }
 
+// NewSenderWithNewSiteAndDeviceWithErrors is the same as NewSenderWithNewSiteAndDeviceWithErrors except rather than
+// passing in a channel to receive errors, a channel is returned by the function. The channel is closed after Sender.Stop is called
+// and all flow has been dispatched
+func NewSenderWithNewSiteAndDeviceWithErrors(siteAndDevice *api.SiteAndDeviceCreate, cfg *Config) (*Sender, <-chan error, error) {
+	client := cfg.client()
+	d, err := client.CreateDeviceAndSite(siteAndDevice)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return cfg.startWithInternalErrors(client, d)
+}
+
+// NewSenderFromDevice returns a Sender for an existing Device
 func NewSenderFromDevice(d *api.Device, errors chan<- error, cfg *Config) (*Sender, error) {
 	client := cfg.client()
 	return cfg.start(client, d, errors)
 }
 
+// NewSenderFromDeviceWithErrors returns a Sender and an error channel for an existing Device
+func NewSenderFromDeviceWithErrors(d *api.Device, cfg *Config) (*Sender, <-chan error, error) {
+	client := cfg.client()
+	return cfg.startWithInternalErrors(client, d)
+}
+
 func lookupdev(dev *api.Device, err error) (*api.Device, error) {
 	if err != nil {
 		switch {

lib_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/kentik/libkflow/flow"
 	metrics2 "github.com/kentik/libkflow/metrics"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
 )
 
@@ -281,6 +282,123 @@ func TestNewSenderWithDeviceNameLeaks(t *testing.T) {
 	goleak.VerifyNone(t, ignore)
 }
 
+func TestNewSenderFromDeviceWithErrors(t *testing.T) {
+	client, server, device, err := test.NewClientServer()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	flowServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(400)
+	}))
+
+	apiurl = server.URL(test.API)
+	flowurl = server.URL(flowServer.URL)
+	metricsurl = server.URL(test.TSDB)
+
+	email = client.Email
+	token = client.Token
+
+	config := libkflow.NewConfig(email, token, "test", "0.0.1")
+	config.OverrideURLs(apiurl, flowurl, metricsurl)
+
+	l := stubLeveledLogger{}
+
+	registry := metrics.NewRegistry()
+	metrics2.StartWithSetConf(registry, &l, metricsurl.String(), email, token, "chf")
+	config.WithRegistry(registry)
+
+	s, errors, err := libkflow.NewSenderFromDeviceWithErrors(device, config)
+
+	require.NotNil(t, s)
+	require.Nil(t, err)
+
+	errorsFromChan := make([]error, 0)
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		for err := range errors {
+			errorsFromChan = append(errorsFromChan, err)
+		}
+		wg.Done()
+	}()
+
+	s.Send(&flow.Flow{
+		TimestampNano: time.Now().UnixNano(),
+	})
+
+	s.Stop(time.Second)
+
+	wg.Wait()
+
+	assert.Len(t, errorsFromChan, 1)
+}
+
+func TestNewSenderWithNewSiteAndDeviceWithErrors(t *testing.T) {
+	client, server, device, err := test.NewClientServer()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	flowServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(400)
+	}))
+
+	apiurl = server.URL(test.API)
+	flowurl = server.URL(flowServer.URL)
+	metricsurl = server.URL(test.TSDB)
+
+	email = client.Email
+	token = client.Token
+
+	config := libkflow.NewConfig(email, token, "test", "0.0.1")
+	config.OverrideURLs(apiurl, flowurl, metricsurl)
+
+	l := stubLeveledLogger{}
+
+	registry := metrics.NewRegistry()
+	metrics2.StartWithSetConf(registry, &l, metricsurl.String(), email, token, "chf")
+	config.WithRegistry(registry)
+
+	s, errors, err := libkflow.NewSenderWithNewSiteAndDeviceWithErrors(&api.SiteAndDeviceCreate{
+		Site: &api.SiteCreate{
+			Title:   "",
+			City:    "",
+			Region:  "",
+			Country: "",
+		},
+		Device: &api.DeviceCreate{
+			AllowNoIP: true,
+			Name:      device.Name,
+		},
+	}, config)
+
+	require.NotNil(t, s)
+	require.Nil(t, err)
+
+	errorsFromChan := make([]error, 0)
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		for err := range errors {
+			errorsFromChan = append(errorsFromChan, err)
+		}
+		wg.Done()
+	}()
+
+	s.Send(&flow.Flow{
+		TimestampNano: time.Now().UnixNano(),
+	})
+
+	s.Stop(time.Second)
+
+	wg.Wait()
+
+	assert.Len(t, errorsFromChan, 1)
+}
+
 func TestNewSenderFromDevice(t *testing.T) {
 	dev, assert := setupLibTest(t)