[Filebeat][CometD] Add CometD input for Salesforce connector

x-pack/filebeat/input/cometd/input.go

@@ -55,8 +55,7 @@ func (in *cometdInput) run() error {
 	for e := range in.msgCh {
 		if e.Failed() {
 			return fmt.Errorf("error collecting events: %w", e.Err)
-		}
-		if !e.Msg.Successful {
+		} else if !e.Msg.Successful {
 			var event event
 			// To handle the last response where the object received was empty
 			if e.Msg.Data.Payload == nil {

x-pack/filebeat/input/cometd/input_test.go

@@ -27,19 +27,15 @@ import (
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
 
-const (
-	firstChannel  = "channel_name1"
-	secondChannel = "channel_name2"
-)
-
 var (
 	serverURL string
+	inputType int
 	called    uint64
 )
 
 func TestInputDone(t *testing.T) {
 	config := mapstr.M{
-		"channel_name":              firstChannel,
+		"channel_name":              "channel_channel",
 		"auth.oauth2.client.id":     "DEMOCLIENTID",
 		"auth.oauth2.client.secret": "DEMOCLIENTSECRET",
 		"auth.oauth2.user":          "salesforce_user",
@@ -68,6 +64,7 @@ func TestMakeEventFailure(t *testing.T) {
 }
 
 func TestSingleInput(t *testing.T) {
+	inputType = 1
 	defer atomic.StoreUint64(&called, 0)
 	eventsCh := make(chan beat.Event)
 	defer close(eventsCh)
@@ -96,7 +93,7 @@ func TestSingleInput(t *testing.T) {
 		"auth.oauth2.password":      "password",
 	}
 
-	r := http.HandlerFunc(oauth2SingleEventHandler)
+	r := http.HandlerFunc(oauth2Handler)
 	server := httptest.NewServer(r)
 	defer server.Close()
 
@@ -117,6 +114,7 @@ func TestSingleInput(t *testing.T) {
 }
 
 func TestInputStop_Wait(t *testing.T) {
+	inputType = 1
 	defer atomic.StoreUint64(&called, 0)
 	eventsCh := make(chan beat.Event)
 	defer close(eventsCh)
@@ -151,7 +149,7 @@ func TestInputStop_Wait(t *testing.T) {
 		"auth.oauth2.password":      "password",
 	}
 
-	r := http.HandlerFunc(oauth2SingleEventHandler)
+	r := http.HandlerFunc(oauth2Handler)
 	server := httptest.NewServer(r)
 	defer server.Close()
 
@@ -170,20 +168,22 @@ func TestInputStop_Wait(t *testing.T) {
 	require.Equal(t, 1, bay.GetConnectedCount())
 
 	var wg sync.WaitGroup
+	var waitForConnections sync.WaitGroup
 	wg.Add(1)
+	waitForConnections.Add(1)
 	go func() {
 		require.Equal(t, 1, bay.GetConnectedCount()) // current open channels count should be 1
 		event := <-eventsCh
 		assertEventMatches(t, expected, event) // wait for single event
 		wg.Done()
 		time.Sleep(100 * time.Millisecond)           // let input.Stop() be executed.
 		require.Equal(t, 0, bay.GetConnectedCount()) // current open channels count should be 0
+		waitForConnections.Done()
 	}()
 
-	time.Sleep(100 * time.Millisecond) // let input.Stop() be executed.
-
 	wg.Wait()
 	input.Wait()
+	waitForConnections.Wait()
 }
 
 func TestStop(t *testing.T) {
@@ -241,6 +241,7 @@ func TestWait(t *testing.T) {
 }
 
 func TestMultiInput(t *testing.T) {
+	inputType = 2
 	defer atomic.StoreUint64(&called, 0)
 	eventsCh := make(chan beat.Event)
 	defer close(eventsCh)
@@ -291,63 +292,28 @@ func TestMultiInput(t *testing.T) {
 	require.NotNil(t, input2)
 
 	require.Equal(t, 0, bay.GetConnectedCount())
-	// run input
-	input1.Run()
-	defer input1.Stop()
-
-	// run input
-	input2.Run()
-	defer input2.Stop()
 
 	got := 0
 	go func() {
-		fmt.Println("start")
+		// run input
+		input1.Run()
+		defer input1.Stop()
+
+		// run input
+		input2.Run()
+		defer input2.Stop()
+
 		for _, event := range []beat.Event{<-eventsCh, <-eventsCh} {
 			assertEventMatches(t, expected1, event)
 			got++
-			fmt.Println(got)
 		}
 	}()
-	time.Sleep(2 * time.Second)
+	time.Sleep(time.Second)
 	if got < 2 {
 		require.NoError(t, fmt.Errorf("not able to get events."))
 	}
 }
 
-func oauth2SingleEventHandler(w http.ResponseWriter, r *http.Request) {
-	w.Header().Set("content-type", "application/json")
-	_ = r.ParseForm()
-	if r.URL.Path == "/token" {
-		response := `{"instance_url": "` + serverURL + `", "expires_in": "60", "access_token": "abcd"}`
-		_, _ = w.Write([]byte(response))
-		return
-	}
-	body, _ := ioutil.ReadAll(r.Body)
-
-	switch string(body) {
-	case `{"channel": "/meta/handshake", "supportedConnectionTypes": ["long-polling"], "version": "1.0"}`:
-		_, _ = w.Write([]byte(`[{"ext":{"replay":true,"payload.format":true},"minimumVersion":"1.0","clientId":"client_id","supportedConnectionTypes":["long-polling"],"channel":"/meta/handshake","version":"1.0","successful":true}]`))
-		return
-	case `{"channel": "/meta/connect", "connectionType": "long-polling", "clientId": "client_id"} `:
-		if called < 1 {
-			atomic.AddUint64(&called, 1)
-			_, _ = w.Write([]byte(`[{"data": {"payload": {"CountryIso": "IN"}, "event": {"replayId":1234}}, "channel": "channel_name"}]`))
-		}
-		return
-	case `{
-								"channel": "/meta/subscribe",
-								"subscription": "channel_name",
-								"clientId": "client_id",
-								"ext": {
-									"replay": {"channel_name": "-1"}
-									}
-								}`:
-		_, _ = w.Write([]byte(`[{"clientId": "client_id", "channel": "/meta/subscribe", "subscription": "channel_name", "successful":true}]`))
-		return
-	default:
-	}
-}
-
 func oauth2Handler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("content-type", "application/json")
 	_ = r.ParseForm()
@@ -363,9 +329,11 @@ func oauth2Handler(w http.ResponseWriter, r *http.Request) {
 		_, _ = w.Write([]byte(`[{"ext":{"replay":true,"payload.format":true},"minimumVersion":"1.0","clientId":"client_id","supportedConnectionTypes":["long-polling"],"channel":"/meta/handshake","version":"1.0","successful":true}]`))
 		return
 	case `{"channel": "/meta/connect", "connectionType": "long-polling", "clientId": "client_id"} `:
-		if called < 2 {
+		if called < uint64(inputType) {
 			atomic.AddUint64(&called, 1)
 			_, _ = w.Write([]byte(`[{"data": {"payload": {"CountryIso": "IN"}, "event": {"replayId":1234}}, "channel": "channel_name"}]`))
+		} else {
+			_, _ = w.Write([]byte(`{}`))
 		}
 		return
 	case `{