chore: enable sfdc source

charts/typhoon/templates/deployment.yaml

@@ -49,6 +49,8 @@ spec:
               value: "{{ .Values.global.image.repository }}/kafkasource-adapter:{{ .Values.global.image.tag | default .Chart.AppVersion }}"
             - name: WEBHOOKSOURCE_IMAGE
               value: "{{ .Values.global.image.repository }}/webhooksource-adapter:{{ .Values.global.image.tag | default .Chart.AppVersion }}"
+            - name: SALESFORCESOURCE_IMAGE
+              value: "{{ .Values.global.image.repository }}/salesforcesource-adapter:{{ .Values.global.image.tag | default .Chart.AppVersion }}"
             # Target adapters
             - name: CLOUDEVENTSTARGET_IMAGE
               value: "{{ .Values.global.image.repository }}/cloudeventstarget-adapter:{{ .Values.global.image.tag | default .Chart.AppVersion }}"

config/500-controller.yaml

@@ -49,6 +49,8 @@ spec:
           value: ko://github.com/zeiss/typhoon/cmd/httppollersource-adapter
         - name: KAFKASOURCE_IMAGE
           value: ko://github.com/zeiss/typhoon/cmd/kafkasource-adapter
+        - name: SALESFORCESOURCE_IMAGE
+          value: ko://github.com/zeiss/typhoon/cmd/salesforcesource-adapter
         - name: WEBHOOKSOURCE_IMAGE
           value: ko://github.com/zeiss/typhoon/cmd/webhooksource-adapter
         - name: CLOUDEVENTSTARGET_IMAGE

pkg/sources/reconciler/salesforcesource/adapter.go

@@ -0,0 +1,93 @@
+package salesforcesource
+
+import (
+	"strconv"
+
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+
+	"knative.dev/eventing/pkg/reconciler/source"
+	"knative.dev/pkg/apis"
+
+	commonv1alpha1 "github.com/zeiss/typhoon/pkg/apis/common/v1alpha1"
+	"github.com/zeiss/typhoon/pkg/apis/sources/v1alpha1"
+	common "github.com/zeiss/typhoon/pkg/reconciler"
+	"github.com/zeiss/typhoon/pkg/reconciler/resource"
+)
+
+const (
+	envSalesforceAuthClientID = "SALESFORCE_AUTH_CLIENT_ID"
+	envSalesforceAuthServer   = "SALESFORCE_AUTH_SERVER"
+	envSalesforceAuthUser     = "SALESFORCE_AUTH_USER"
+	envSalesforceAuthCertKey  = "SALESFORCE_AUTH_CERT_KEY"
+	envSalesforceAPIVersion   = "SALESFORCE_API_VERSION"
+	envSalesforceChannel      = "SALESFORCE_SUBCRIPTION_CHANNEL"
+	envSalesforceReplayID     = "SALESFORCE_SUBCRIPTION_REPLAY_ID"
+)
+
+// adapterConfig contains properties used to configure the source's adapter.
+// These are automatically populated by envconfig.
+type adapterConfig struct {
+	// Container image
+	Image string `default:"gcr.io/triggermesh/salesforcesource-adapter"`
+	// Configuration accessor for logging/metrics/tracing
+	configs source.ConfigAccessor
+}
+
+// Verify that Reconciler implements common.AdapterBuilder.
+var _ common.AdapterBuilder[*appsv1.Deployment] = (*Reconciler)(nil)
+
+// BuildAdapter implements common.AdapterBuilder.
+func (r *Reconciler) BuildAdapter(src commonv1alpha1.Reconcilable, sinkURI *apis.URL) (*appsv1.Deployment, error) {
+	typedSrc := src.(*v1alpha1.SalesforceSource)
+
+	return common.NewAdapterDeployment(src, sinkURI,
+		resource.Image(r.adapterCfg.Image),
+
+		resource.EnvVars(MakeAppEnv(typedSrc)...),
+		resource.EnvVars(r.adapterCfg.configs.ToEnvVars()...),
+	), nil
+}
+
+// MakeAppEnv extracts environment variables from the object.
+// Exported to be used in external tools for local test environments.
+func MakeAppEnv(o *v1alpha1.SalesforceSource) []corev1.EnvVar {
+	appEnv := []corev1.EnvVar{
+		{
+			Name:  envSalesforceAuthClientID,
+			Value: o.Spec.Auth.ClientID,
+		},
+		{
+			Name:  envSalesforceAuthServer,
+			Value: o.Spec.Auth.Server,
+		},
+		{
+			Name:  envSalesforceAuthUser,
+			Value: o.Spec.Auth.User,
+		},
+		{
+			Name:  envSalesforceChannel,
+			Value: o.Spec.Subscription.Channel,
+		},
+	}
+
+	appEnv = common.MaybeAppendValueFromEnvVar(appEnv,
+		envSalesforceAuthCertKey, o.Spec.Auth.CertKey,
+	)
+
+	if o.Spec.Subscription.ReplayID != nil {
+		appEnv = append(appEnv, corev1.EnvVar{
+			Name:  envSalesforceReplayID,
+			Value: strconv.Itoa(*o.Spec.Subscription.ReplayID),
+		})
+	}
+
+	if o.Spec.APIVersion != nil {
+		appEnv = append(appEnv, corev1.EnvVar{
+			Name:  envSalesforceAPIVersion,
+			Value: *o.Spec.APIVersion,
+		})
+	}
+
+	return appEnv
+}

pkg/sources/reconciler/salesforcesource/controller.go

@@ -0,0 +1,52 @@
+package salesforcesource
+
+import (
+	"context"
+
+	"github.com/kelseyhightower/envconfig"
+
+	"knative.dev/eventing/pkg/reconciler/source"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/controller"
+
+	"github.com/zeiss/typhoon/pkg/apis/sources/v1alpha1"
+	informerv1alpha1 "github.com/zeiss/typhoon/pkg/client/generated/injection/informers/sources/v1alpha1/salesforcesource"
+	reconcilerv1alpha1 "github.com/zeiss/typhoon/pkg/client/generated/injection/reconciler/sources/v1alpha1/salesforcesource"
+	common "github.com/zeiss/typhoon/pkg/reconciler"
+)
+
+// NewController creates a Reconciler for the event source and returns the result of NewImpl.
+func NewController(
+	ctx context.Context,
+	cmw configmap.Watcher,
+) *controller.Impl {
+
+	typ := (*v1alpha1.SalesforceSource)(nil)
+	app := common.ComponentName(typ)
+
+	// Calling envconfig.Process() with a prefix appends that prefix
+	// (uppercased) to the Go field name, e.g. MYSOURCE_IMAGE.
+	adapterCfg := &adapterConfig{
+		configs: source.WatchConfigurations(ctx, app, cmw),
+	}
+	envconfig.MustProcess(app, adapterCfg)
+
+	informer := informerv1alpha1.Get(ctx)
+
+	r := &Reconciler{
+		adapterCfg: adapterCfg,
+	}
+	impl := reconcilerv1alpha1.NewImpl(ctx, r)
+
+	r.base = common.NewGenericDeploymentReconciler[*v1alpha1.SalesforceSource](
+		ctx,
+		typ.GetGroupVersionKind(),
+		impl.Tracker,
+		impl.EnqueueControllerOf,
+		informer.Lister().SalesforceSources,
+	)
+
+	informer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
+
+	return impl
+}

pkg/sources/reconciler/salesforcesource/reconciler.go

@@ -0,0 +1,30 @@
+package salesforcesource
+
+import (
+	"context"
+
+	"knative.dev/pkg/reconciler"
+
+	commonv1alpha1 "github.com/zeiss/typhoon/pkg/apis/common/v1alpha1"
+	"github.com/zeiss/typhoon/pkg/apis/sources/v1alpha1"
+	reconcilerv1alpha1 "github.com/zeiss/typhoon/pkg/client/generated/injection/reconciler/sources/v1alpha1/salesforcesource"
+	listersv1alpha1 "github.com/zeiss/typhoon/pkg/client/generated/listers/sources/v1alpha1"
+	common "github.com/zeiss/typhoon/pkg/reconciler"
+)
+
+// Reconciler implements controller.Reconciler for the event source type.
+type Reconciler struct {
+	base       common.GenericDeploymentReconciler[*v1alpha1.SalesforceSource, listersv1alpha1.SalesforceSourceNamespaceLister]
+	adapterCfg *adapterConfig
+}
+
+// Check that our Reconciler implements Interface
+var _ reconcilerv1alpha1.Interface = (*Reconciler)(nil)
+
+// ReconcileKind implements Interface.ReconcileKind.
+func (r *Reconciler) ReconcileKind(ctx context.Context, src *v1alpha1.SalesforceSource) reconciler.Event {
+	// inject source into context for usage in reconciliation logic
+	ctx = commonv1alpha1.WithReconcilable(ctx, src)
+
+	return r.base.ReconcileAdapter(ctx, r)
+}