valid gcp-pubsub gateway type plus validation of parameter (#492)

examples/gateways/gcp-pubsub.yaml

@@ -7,7 +7,7 @@ metadata:
     gateways.argoproj.io/gateway-controller-instanceid: argo-events
 spec:
   replica: 1
-  type: gcp-pubsub
+  type: pubsub
   eventSourceRef:
     name: gcp-pubsub-event-source
   template:

gateways/client/event-source_test.go

@@ -140,7 +140,8 @@ func TestInitEventSourceContexts(t *testing.T) {
 		fmt.Println("server is stopped")
 	}()
 
-	contexts := gatewayContext.initEventSourceContexts(eventSource)
+	contexts, err := gatewayContext.initEventSourceContexts(eventSource)
+	assert.NoError(t, err)
 	assert.NotNil(t, contexts)
 	for _, esContext := range contexts {
 		assert.Equal(t, "first-webhook", esContext.source.Name)

gateways/client/event-sources.go

@@ -19,6 +19,8 @@ package main
 import (
 	"context"
 	"fmt"
+	"io"
+
 	"github.com/argoproj/argo-events/common"
 	"github.com/argoproj/argo-events/gateways"
 	apicommon "github.com/argoproj/argo-events/pkg/apis/common"
@@ -28,7 +30,6 @@ import (
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/connectivity"
-	"io"
 )
 
 // populateEventSourceContexts sets up the contexts for event sources
@@ -266,7 +267,10 @@ func (gatewayContext *GatewayContext) deactivateEventSources(eventSourceNames []
 
 // syncEventSources syncs active event-sources and the updated ones
 func (gatewayContext *GatewayContext) syncEventSources(eventSource *eventSourceV1Alpha1.EventSource) error {
-	eventSourceContexts := gatewayContext.initEventSourceContexts(eventSource)
+	eventSourceContexts, err := gatewayContext.initEventSourceContexts(eventSource)
+	if err != nil {
+		return err
+	}
 
 	staleEventSources, newEventSources := gatewayContext.diffEventSources(eventSourceContexts)
 	gatewayContext.logger.WithField(common.LabelEventSource, staleEventSources).Infoln("deleted event sources")
@@ -284,8 +288,11 @@ func (gatewayContext *GatewayContext) syncEventSources(eventSource *eventSourceV
 }
 
 // initEventSourceContext creates an internal representation of event sources.
-func (gatewayContext *GatewayContext) initEventSourceContexts(eventSource *eventSourceV1Alpha1.EventSource) map[string]*EventSourceContext {
+// It returns an error if the Gateway is set in such a way
+// that it wouldn't pick up any known Event Source.
+func (gatewayContext *GatewayContext) initEventSourceContexts(eventSource *eventSourceV1Alpha1.EventSource) (map[string]*EventSourceContext, error) {
 	eventSourceContexts := make(map[string]*EventSourceContext)
+	var err error
 
 	switch gatewayContext.gateway.Spec.Type {
 	case apicommon.SNSEvent:
@@ -380,7 +387,9 @@ func (gatewayContext *GatewayContext) initEventSourceContexts(eventSource *event
 		for key, value := range eventSource.Spec.Generic {
 			gatewayContext.populateEventSourceContexts(key, value, eventSourceContexts)
 		}
+	default:
+		err = fmt.Errorf("gateway with type %s is invalid", gatewayContext.gateway.Spec.Type)
 	}
 
-	return eventSourceContexts
+	return eventSourceContexts, err
 }