Repository: triggermesh/aws-event-sources
Pull request #158 (2020-06-12 20:13:43)

Refactor SNS source
Controller:
* switch source deployment mode from Deployment to Knative Service
* update all tests and internal libraries to handle Knative Services as well as Deployments

Adapter:
* subscribe to the adapter's public URL
* allow configuration of subscription attributes
* pass unmodified notification payload in CloudEvents
* handle termination

#157 is causing serious limitations to this source unfortunately.


Patch:
diff --git a/Makefile b/Makefile
index d9b95ae..03fbf53 100644
--- a/Makefile
+++ b/Makefile
@@ -1,6 +1,6 @@
 KREPO              = aws-event-sources
 KREPO_DESC         = Triggermesh AWS event sources
-COMMANDS           = aws-event-sources-controller awscodecommitsource awscognitosource awscognitouserpoolsource awsdynamodbsource awskinesissource awssqssource
+COMMANDS           = aws-event-sources-controller awscodecommitsource awscognitoidentitysource awscognitouserpoolsource awsdynamodbsource awskinesissource awssqssource
 
 TARGETS           ?= linux/amd64
 
diff --git a/README.md b/README.md
index 95e0814..a5bb0d0 100644
--- a/README.md
+++ b/README.md
@@ -10,15 +10,16 @@
 
 ## List of event sources
 
-| AWS service | Support level |
-|-------------|---------------|
-| CodeCommit  | alpha         |
-| Cognito     | alpha         |
-| DynamoDB    | alpha         |
-| IoT         | alpha/WIP     |
-| Kinesis     | alpha         |
-| SNS         | alpha         |
-| SQS         | alpha         |
+| AWS service     | Support level |
+|-----------------|---------------|
+| CodeCommit      | alpha         |
+| CognitoUserPool | alpha         |
+| CognitoIdentity | alpha         |
+| DynamoDB        | alpha         |
+| IoT             | alpha/WIP     |
+| Kinesis         | alpha         |
+| SNS             | alpha         |
+| SQS             | alpha         |
 
 ## Usage
 
diff --git a/cmd/aws-event-sources-controller/main.go b/cmd/aws-event-sources-controller/main.go
index 661f4ee..5a3aa4e 100644
--- a/cmd/aws-event-sources-controller/main.go
+++ b/cmd/aws-event-sources-controller/main.go
@@ -20,7 +20,7 @@ import (
 	"knative.dev/pkg/injection/sharedmain"
 
 	"github.com/triggermesh/aws-event-sources/pkg/reconciler/awscodecommitsource"
-	"github.com/triggermesh/aws-event-sources/pkg/reconciler/awscognitosource"
+	"github.com/triggermesh/aws-event-sources/pkg/reconciler/awscognitoidentitysource"
 	"github.com/triggermesh/aws-event-sources/pkg/reconciler/awscognitouserpoolsource"
 	"github.com/triggermesh/aws-event-sources/pkg/reconciler/awsdynamodbsource"
 	"github.com/triggermesh/aws-event-sources/pkg/reconciler/awsiotsource"
@@ -32,7 +32,7 @@ import (
 func main() {
 	sharedmain.Main("aws-event-sources-controller",
 		awscodecommitsource.NewController,
-		awscognitosource.NewController,
+		awscognitoidentitysource.NewController,
 		awscognitouserpoolsource.NewController,
 		awsdynamodbsource.NewController,
 		awsiotsource.NewController,
diff --git a/cmd/awscognitosource/Dockerfile b/cmd/awscognitoidentitysource/Dockerfile
similarity index 64%
rename from cmd/awscognitosource/Dockerfile
rename to cmd/awscognitoidentitysource/Dockerfile
index 756f5a1..e84a648 100644
--- a/cmd/awscognitosource/Dockerfile
+++ b/cmd/awscognitoidentitysource/Dockerfile
@@ -4,13 +4,13 @@ ENV CGO_ENABLED 0
 ENV GOOS linux
 ENV GOARCH amd64
 
-WORKDIR /go/src/awscognitosource
+WORKDIR /go/src/awscognitoidentitysource
 
 COPY go.mod go.sum ./
 RUN go mod download
 
 COPY . .
-RUN BIN_OUTPUT_DIR=/bin make awscognitosource && \
+RUN BIN_OUTPUT_DIR=/bin make awscognitoidentitysource && \
     mkdir /kodata && \
     mv .git/* /kodata/ && \
     rm -rf ${GOPATH} && \
@@ -18,19 +18,19 @@ RUN BIN_OUTPUT_DIR=/bin make awscognitosource && \
 
 FROM registry.access.redhat.com/ubi8/ubi-minimal
 
-LABEL name "Triggermesh AWS Cognito Event Source"
+LABEL name "Triggermesh AWS Cognito Identity Pool Event Source"
 LABEL vendor "Triggermesh"
 LABEL version "v0.1.0"
 LABEL release "1"
-LABEL summary "The Triggermesh Cognito Source"
-LABEL description "This is the Triggermesh Knative Event Source for AWS Cognito"
+LABEL summary "The Triggermesh Cognito Identity Pool Source"
+LABEL description "This is the Triggermesh Knative Event Source for AWS Cognito Identity Pool"
 
 # Emulate ko builds
 # https://github.com/google/ko/blob/v0.5.0/README.md#including-static-assets
 ENV KO_DATA_PATH /kodata
 
 COPY --from=builder /kodata/ ${KO_DATA_PATH}/
-COPY --from=builder /bin/awscognitosource /
+COPY --from=builder /bin/awscognitoidentitysource /
 COPY licenses/ /licenses/
 
-ENTRYPOINT ["/awscognitosource"]
+ENTRYPOINT ["/awscognitoidentitysource"]
diff --git a/cmd/awscognitosource/README.md b/cmd/awscognitoidentitysource/README.md
similarity index 82%
rename from cmd/awscognitosource/README.md
rename to cmd/awscognitoidentitysource/README.md
index 355a2d4..7c71b0a 100644
--- a/cmd/awscognitosource/README.md
+++ b/cmd/awscognitoidentitysource/README.md
@@ -7,7 +7,7 @@ sink.
 
 1. [Prerequisites](#prerequisites)
 1. [Deployment to Kubernetes](#deployment-to-kubernetes)
-   * [As a AWSCognitoSource object](#as-a-awscognitosource-object)
+   * [As a AWSCognitoIdentitySource object](#as-a-awscognitoidentitysource-object)
    * [As a ContainerSource object](#as-a-containersource-object)
    * [As a Deployment object bound by a SinkBinding](#as-a-deployment-object-bound-by-a-sinkbinding)
 1. [Running locally](#running-locally)
@@ -24,7 +24,7 @@ sink.
 
 The _AWS Cognito event source_ can be deployed to Kubernetes in different manners:
 
-* As an `AWSCognitoSource` object, to a cluster where the TriggerMesh _AWS Sources Controller_ is running.
+* As an `AWSCognitoIdentitySource` object, to a cluster where the TriggerMesh _AWS Sources Controller_ is running.
 * As a Knative `ContainerSource`, to any cluster running Knative Eventing.
 
 > :information_source: The sample manifests below reference AWS credentials (Access Key) from a Kubernetes Secret object
@@ -39,14 +39,14 @@ The _AWS Cognito event source_ can be deployed to Kubernetes in different manner
 > Alternatively, credentials can be used as literal strings instead of references to Kubernetes Secrets by replacing
 > `valueFrom` attributes with `value` inside API objects' manifests.
 
-### As a AWSCognitoSource object
+### As a AWSCognitoIdentitySource object
 
-Copy the sample manifest from `config/samples/awscognitosource.yaml` and replace the pre-filled `spec` attributes
-with the values corresponding to your _AWS Cognito_ identity pool. Then, create that `AWSCognitoSource` object in
+Copy the sample manifest from `config/samples/awscognitoidentitysource.yaml` and replace the pre-filled `spec` attributes
+with the values corresponding to your _AWS Cognito_ identity pool. Then, create that `AWSCognitoIdentitySource` object in
 your Kubernetes cluster:
 
 ```console
-$ kubectl -n <my_namespace> create -f my-awscognitosource.yaml
+$ kubectl -n <my_namespace> create -f my-awscognitoidentitysource.yaml
 ```
 
 ### As a ContainerSource object
@@ -81,16 +81,16 @@ Ensure the following environment variables are exported to your current shell's
 export ARN=<arn_of_my_identity_pool>
 export AWS_ACCESS_KEY_ID=<my_key_id>
 export AWS_SECRET_ACCESS_KEY=<my_secret_key>
-export NAME=my-awscognitosource
+export NAME=my-awscognitoidentitysource
 export NAMESPACE=default
 export K_LOGGING_CONFIG=''
-export K_METRICS_CONFIG='{"domain":"triggermesh.io/sources", "component":"awscognitosource", "configMap":{}}'
+export K_METRICS_CONFIG='{"domain":"triggermesh.io/sources", "component":"awscognitoidentitysource", "configMap":{}}'
 ```
 
 Then, run the event source with:
 
 ```console
-$ go run ./cmd/awscognitosource
+$ go run ./cmd/awscognitoidentitysource
 ```
 
 ### In a Docker container
@@ -102,11 +102,11 @@ $ docker run --rm \
   -e ARN=<arn_of_my_identity_pool> \
   -e AWS_ACCESS_KEY_ID=<my_key_id> \
   -e AWS_SECRET_ACCESS_KEY=<my_secret_key> \
-  -e NAME=my-awscognitosource \
+  -e NAME=my-awscognitoidentitysource \
   -e NAMESPACE=default \
   -e K_LOGGING_CONFIG='' \
-  -e K_METRICS_CONFIG='{"domain":"triggermesh.io/sources", "component":"awscognitosource", "configMap":{}}' \
-  gcr.io/triggermesh/awscognitosource:latest
+  -e K_METRICS_CONFIG='{"domain":"triggermesh.io/sources", "component":"awscognitoidentitysource", "configMap":{}}' \
+  gcr.io/triggermesh/awscognitoidentitysource:latest
 ```
 
 [doc-accesskey]: https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html#access-keys-and-secret-access-keys
diff --git a/cmd/awscognitosource/kodata/HEAD b/cmd/awscognitoidentitysource/kodata/HEAD
similarity index 100%
rename from cmd/awscognitosource/kodata/HEAD
rename to cmd/awscognitoidentitysource/kodata/HEAD
diff --git a/cmd/awscognitosource/kodata/refs b/cmd/awscognitoidentitysource/kodata/refs
similarity index 100%
rename from cmd/awscognitosource/kodata/refs
rename to cmd/awscognitoidentitysource/kodata/refs
diff --git a/cmd/awscognitosource/main.go b/cmd/awscognitoidentitysource/main.go
similarity index 84%
rename from cmd/awscognitosource/main.go
rename to cmd/awscognitoidentitysource/main.go
index 9120d09..19c9cd0 100644
--- a/cmd/awscognitosource/main.go
+++ b/cmd/awscognitoidentitysource/main.go
@@ -19,9 +19,9 @@ package main
 import (
 	"knative.dev/eventing/pkg/adapter/v2"
 
-	"github.com/triggermesh/aws-event-sources/pkg/adapter/awscognitosource"
+	"github.com/triggermesh/aws-event-sources/pkg/adapter/awscognitoidentitysource"
 )
 
 func main() {
-	adapter.Main("awscognitosource", awscognitosource.NewEnvConfig, awscognitosource.NewAdapter)
+	adapter.Main("awscognitoidentitysource", awscognitoidentitysource.NewEnvConfig, awscognitoidentitysource.NewAdapter)
 }
diff --git a/cmd/awscognitouserpoolsource/.gitignore b/cmd/awscognitouserpoolsource/.gitignore
deleted file mode 100644
index 1323f98..0000000
--- a/cmd/awscognitouserpoolsource/.gitignore
+++ /dev/null
@@ -1,1 +0,0 @@
-awscognitouserpoolsource
\ No newline at end of file
diff --git a/cmd/awssnssource/README.md b/cmd/awssnssource/README.md
index 03d89d5..c0d4b52 100644
--- a/cmd/awssnssource/README.md
+++ b/cmd/awssnssource/README.md
@@ -2,17 +2,8 @@
 
 This event source subscribes to messages from a AWS SNS topic and sends them as CloudEvents to an arbitrary event sink.
 
-<!--
-This source expects external-dns of some flavor (https://github.com/kubernetes-sigs/external-dns) to be installed in the
-k8s cluster. If you look at the deployment.yaml file, you'll see that a service is created that has a hostname
-annotation. This hostname must be of the form `sns.$TOPIC_NAME.$CHANNEL_NAME.$NAMESPACE.$DOMAIN`. Domain can be anything
-you like. I used a subdomain in my personal env `sources.rsmitty.cloud`. These same vars are passed into the SNS
-deployment as env variables.
-
-Upon starting, a background task is launched before running the webserver. This background task attempts to subscribe to
-the SNS topic mentioned. It will wait until the hostname above resolves in DNS before attempting to do so. Once
-subscription has occurred, events are pushed like other sources.
--->
+Each instance of the SNS source is backed by a Knative Service that exposes a unique public HTTP(S) endpoint. This
+endpoint is used to subscribe to the desired SNS topic on behalf of the user.
 
 ## Contents
 
@@ -30,7 +21,6 @@ subscription has occurred, events are pushed like other sources.
 * Register an AWS account
 * Create an [Access Key][doc-accesskey] in your AWS IAM dashboard.
 * Create a [SNS topic][doc-sns].
-* Create a [SNS subscription][doc-sns].
 
 ## Deployment to Kubernetes
 
@@ -90,6 +80,7 @@ Ensure the following environment variables are exported to your current shell's
 
 ```sh
 export ARN=<arn_of_my_sns_topic>
+export PUBLIC_URL=<public_source_url>
 export AWS_ACCESS_KEY_ID=<my_key_id>
 export AWS_SECRET_ACCESS_KEY=<my_secret_key>
 export NAME=my-awssnssource
@@ -111,6 +102,7 @@ Using one of TriggerMesh's release images:
 ```console
 $ docker run --rm \
   -e ARN=<arn_of_my_sns_topic> \
+  -e PUBLIC_URL=<public_source_url> \
   -e AWS_ACCESS_KEY_ID=<my_key_id> \
   -e AWS_SECRET_ACCESS_KEY=<my_secret_key> \
   -e NAME=my-awssnssource \
@@ -121,4 +113,4 @@ $ docker run --rm \
 ```
 
 [doc-accesskey]: https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html#access-keys-and-secret-access-keys
-[doc-sns-topic]: https://docs.aws.amazon.com/sns/latest/dg/sns-getting-started.html
+[doc-sns]: https://docs.aws.amazon.com/sns/latest/dg/sns-getting-started.html
diff --git a/config/200-clusterroles.yaml b/config/200-clusterroles.yaml
index 80a3d0f..911c100 100644
--- a/config/200-clusterroles.yaml
+++ b/config/200-clusterroles.yaml
@@ -52,7 +52,7 @@ rules:
   - sources.triggermesh.io
   resources:
   - awscodecommitsources
-  - awscognitosources
+  - awscognitoidentitysources
   - awscognitouserpoolsources
   - awsdynamodbsources
   - awsiotsources
@@ -68,7 +68,7 @@ rules:
   - sources.triggermesh.io
   resources:
   - awscodecommitsources/status
-  - awscognitosources/status
+  - awscognitoidentitysources/status
   - awscognitouserpoolsources/status
   - awsdynamodbsources/status
   - awsiotsources/status
@@ -134,7 +134,7 @@ rules:
   - sources.triggermesh.io
   resources:
   - awscodecommitsources
-  - awscognitosources
+  - awscognitoidentitysources
   - awscognitouserpoolsources
   - awsdynamodbsources
   - awsiotsources
diff --git a/config/300-awscognito.yaml b/config/300-awscognitoidentity.yaml
similarity index 96%
rename from config/300-awscognito.yaml
rename to config/300-awscognitoidentity.yaml
index 10a54dc..10dfde7 100644
--- a/config/300-awscognito.yaml
+++ b/config/300-awscognitoidentity.yaml
@@ -15,7 +15,7 @@
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
-  name: awscognitosources.sources.triggermesh.io
+  name: awscognitoidentitysources.sources.triggermesh.io
   labels:
     eventing.knative.dev/source: 'true'
     duck.knative.dev/source: 'true'
@@ -23,14 +23,14 @@ metadata:
   annotations:
     registry.knative.dev/eventTypes: |
       [
-        { "type": "com.amazon.cognito.sync_trigger" }
+        { "type": "com.amazon.cognito-identity.sync_trigger" }
       ]
 spec:
   group: sources.triggermesh.io
   scope: Namespaced
   names:
-    kind: AWSCognitoSource
-    plural: awscognitosources
+    kind: AWSCognitoIdentitySource
+    plural: awscognitoidentitysources
     categories:
     - all
     - knative
diff --git a/config/300-awssns.yaml b/config/300-awssns.yaml
index 607af0b..764d8d1 100644
--- a/config/300-awssns.yaml
+++ b/config/300-awssns.yaml
@@ -52,6 +52,25 @@ spec:
               arn:
                 type: string
                 pattern: '^arn:aws(-cn|-us-gov)?:sns:[a-z]{2}(-gov)?-[a-z]+-\d:\d{12}:.+$'
+              subscriptionAttributes:
+                type: object
+                properties:
+                  DeliveryPolicy:
+                    type: string
+                    format: json
+                    nullable: true
+                  FilterPolicy:
+                    type: string
+                    format: json
+                    nullable: true
+                  RawMessageDelivery:
+                    type: string
+                    format: json
+                    nullable: true
+                  RedrivePolicy:
+                    type: string
+                    format: json
+                    nullable: true
               credentials:
                 type: object
                 properties:
@@ -157,6 +176,11 @@ spec:
                   required:
                   - type
                   - status
+              address:
+                type: object
+                properties:
+                  url:
+                    type: string
     additionalPrinterColumns:
     - name: Ready
       type: string
@@ -164,6 +188,9 @@ spec:
     - name: Reason
       type: string
       jsonPath: .status.conditions[?(@.type=='Ready')].reason
+    - name: URL
+      type: string
+      jsonPath: .status.address.url
     - name: Sink
       type: string
       jsonPath: .status.sinkUri
diff --git a/config/500-controller.yaml b/config/500-controller.yaml
index 3753692..f81db9e 100644
--- a/config/500-controller.yaml
+++ b/config/500-controller.yaml
@@ -54,11 +54,16 @@ spec:
           value: config-observability
         - name: METRICS_DOMAIN
           value: triggermesh.io/sources
+        # Global settings
+        - name: KNATIVE_DOMAIN
+          value: example.com
+        - name: KNATIVE_URL_SCHEME
+          value: http
         # Source adapters
         - name: AWSCODECOMMITSOURCE_IMAGE
           value: ko://github.com/triggermesh/aws-event-sources/cmd/awscodecommitsource
-        - name: AWSCOGNITOSOURCE_IMAGE
-          value: ko://github.com/triggermesh/aws-event-sources/cmd/awscognitosource
+        - name: AWSCOGNITOIDENTITYSOURCE_IMAGE
+          value: ko://github.com/triggermesh/aws-event-sources/cmd/awscognitoidentitysource
         - name: AWSCOGNITOUSERPOOLSOURCE_IMAGE
           value: ko://github.com/triggermesh/aws-event-sources/cmd/awscognitouserpoolsource
         - name: AWSDYNAMODBSOURCE_IMAGE
diff --git a/config/samples/awscognito-containersource.yaml b/config/samples/awscognitoidentity-containersource.yaml
similarity index 93%
rename from config/samples/awscognito-containersource.yaml
rename to config/samples/awscognitoidentity-containersource.yaml
index b614e0b..3dcff73 100644
--- a/config/samples/awscognito-containersource.yaml
+++ b/config/samples/awscognitoidentity-containersource.yaml
@@ -23,7 +23,7 @@ spec:
     spec:
       containers:
       - name: source
-        image: gcr.io/triggermesh/awscognitosource:latest
+        image: gcr.io/triggermesh/awscognitoidentitysource:latest
 
         env:
 
@@ -56,7 +56,7 @@ spec:
           value: |
             {
               "domain": "triggermesh.io/sources",
-              "component": "awscognitosource",
+              "component": "awscognitoidentitysource",
               "configMap": {}
             }
 
diff --git a/config/samples/awscognito-sinkbinding.yaml b/config/samples/awscognitoidentity-sinkbinding.yaml
similarity index 94%
rename from config/samples/awscognito-sinkbinding.yaml
rename to config/samples/awscognitoidentity-sinkbinding.yaml
index a1867ac..128a3fd 100644
--- a/config/samples/awscognito-sinkbinding.yaml
+++ b/config/samples/awscognitoidentity-sinkbinding.yaml
@@ -49,7 +49,7 @@ spec:
     spec:
       containers:
       - name: source
-        image: gcr.io/triggermesh/awscognitosource:latest
+        image: gcr.io/triggermesh/awscognitoidentitysource:latest
 
         env:
 
@@ -82,6 +82,6 @@ spec:
           value: |
             {
               "domain": "triggermesh.io/sources",
-              "component": "awscognitosource",
+              "component": "awscognitoidentitysource",
               "configMap": {}
             }
diff --git a/config/samples/awscognitosource.yaml b/config/samples/awscognitoidentitysource.yaml
similarity index 93%
rename from config/samples/awscognitosource.yaml
rename to config/samples/awscognitoidentitysource.yaml
index d7d8fc0..5fe0464 100644
--- a/config/samples/awscognitosource.yaml
+++ b/config/samples/awscognitoidentitysource.yaml
@@ -12,10 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# Sample AWSCognitoSource object.
+# Sample AWSCognitoIdentitySource object.
 
 apiVersion: sources.triggermesh.io/v1alpha1
-kind: AWSCognitoSource
+kind: AWSCognitoIdentitySource
 metadata:
   name: sample
 spec:
diff --git a/config/samples/awscognitouserpoolsource.yaml b/config/samples/awscognitouserpoolsource.yaml
index edc4714..1e7917f 100644
--- a/config/samples/awscognitouserpoolsource.yaml
+++ b/config/samples/awscognitouserpoolsource.yaml
@@ -36,4 +36,3 @@ spec:
       apiVersion: eventing.knative.dev/v1beta1
       kind: Broker
       name: default
-
diff --git a/config/samples/awssnssource.yaml b/config/samples/awssnssource.yaml
index 815e340..b9a5b82 100644
--- a/config/samples/awssnssource.yaml
+++ b/config/samples/awssnssource.yaml
@@ -21,6 +21,17 @@ metadata:
 spec:
   arn: arn:aws:sns:us-west-2:123456789012:triggermeshtest
 
+  # For a list of supported subscription attributes, please refer to the following resources:
+  #  * https://docs.aws.amazon.com/sns/latest/api/API_SetSubscriptionAttributes.html
+  #  * https://docs.aws.amazon.com/sns/latest/dg/sns-how-it-works.html
+  subscriptionAttributes:
+    DeliveryPolicy: |
+      {
+        "healthyRetryPolicy": {
+          "numRetries": 5
+        }
+      }
+
   credentials:
     accessKeyID:
       valueFromSecret:
diff --git a/go.mod b/go.mod
index 343f6d2..4650f2b 100644
--- a/go.mod
+++ b/go.mod
@@ -26,4 +26,5 @@ require (
 	k8s.io/utils v0.0.0-20200327001022-6496210b90e8 // indirect
 	knative.dev/eventing v0.14.1-0.20200512233457-09a6267a48c8
 	knative.dev/pkg v0.0.0-20200519155757-14eb3ae3a5a7
+	knative.dev/serving v0.14.1
 )
diff --git a/go.sum b/go.sum
index cc86de2..f1ec44e 100644
--- a/go.sum
+++ b/go.sum
@@ -465,6 +465,7 @@ github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-containerregistry v0.0.0-20191010200024-a3d713f9b7f8/go.mod h1:KyKXa9ciM8+lgMXwOVsXi7UxGrsf9mM61Mzs+xKUrKE=
 github.com/google/go-containerregistry v0.0.0-20200115214256-379933c9c22b/go.mod h1:Wtl/v6YdQxv397EREtzwgd9+Ud7Q5D8XMbi3Zazgkrs=
+github.com/google/go-containerregistry v0.0.0-20200123184029-53ce695e4179 h1:wFBYu1QOSE+sgYeX2jtZUldOgLUebWYm/thF0Et7U8o=
 github.com/google/go-containerregistry v0.0.0-20200123184029-53ce695e4179/go.mod h1:Wtl/v6YdQxv397EREtzwgd9+Ud7Q5D8XMbi3Zazgkrs=
 github.com/google/go-github v17.0.0+incompatible/go.mod h1:zLgOLi98H3fifZn+44m+umXrS52loVEgC2AApnigrVQ=
 github.com/google/go-github/v27 v27.0.6/go.mod h1:/0Gr8pJ55COkmv+S/yPKCczSkUPIM/LnFyubufRNIS0=
@@ -1480,6 +1481,8 @@ knative.dev/pkg v0.0.0-20200509234445-b52862b1b3ea/go.mod h1:VR1a7CeToIFubOhOlip
 knative.dev/pkg v0.0.0-20200515002500-16d7b963416f/go.mod h1:tMOHGbxtRz8zYFGEGpV/bpoTEM1o89MwYFC4YJXl3GY=
 knative.dev/pkg v0.0.0-20200519155757-14eb3ae3a5a7 h1:9S2r59HZJF9nKvoRLg5zJzx6XpVlVyvVRqz/C/h6h2s=
 knative.dev/pkg v0.0.0-20200519155757-14eb3ae3a5a7/go.mod h1:QgNZTxnwpB/oSpNcfnLVlw+WpEwwyKAvJlvR3hgeltA=
+knative.dev/serving v0.14.1 h1:Fw702AUzuHVF+creWcTCKq0PsQYEDXixp3mO8EH0a/A=
+knative.dev/serving v0.14.1/go.mod h1:x2n255JS2XBI39tmjZ8CwTxIf9EKNMCrkVuiOttLRm0=
 knative.dev/test-infra v0.0.0-20200407185800-1b88cb3b45a5/go.mod h1:xcdUkMJrLlBswIZqL5zCuBFOC22WIPMQoVX1L35i0vQ=
 knative.dev/test-infra v0.0.0-20200505052144-5ea2f705bb55/go.mod h1:WqF1Azka+FxPZ20keR2zCNtiQA1MP9ZB4BH4HuI+SIU=
 knative.dev/test-infra v0.0.0-20200508015845-8d7d46a46176/go.mod h1:aMif0KXL4g19YCYwsy4Ocjjz5xgPlseYV+B95Oo4JGE=
diff --git a/pkg/adapter/awscognitosource/adapter.go b/pkg/adapter/awscognitoidentitysource/adapter.go
similarity index 97%
rename from pkg/adapter/awscognitosource/adapter.go
rename to pkg/adapter/awscognitoidentitysource/adapter.go
index f47d5f7..cdb1da9 100644
--- a/pkg/adapter/awscognitosource/adapter.go
+++ b/pkg/adapter/awscognitoidentitysource/adapter.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	"context"
@@ -85,7 +85,7 @@ func NewAdapter(ctx context.Context, envAcc pkgadapter.EnvConfigAccessor, ceClie
 		ceClient:          ceClient,
 
 		arn:            arn,
-		identityPoolID: common.MustParseCognitoResource(arn.Resource),
+		identityPoolID: common.MustParseCognitoIdentityResource(arn.Resource),
 	}
 }
 
@@ -207,7 +207,7 @@ func (a *adapter) getRecords(dataset *cognitosync.Dataset) ([]*cognitosync.Recor
 func (a *adapter) sendCognitoEvent(dataset *cognitosync.Dataset, records []*cognitosync.Record) error {
 	a.logger.Info("Processing Dataset: ", *dataset.DatasetName)
 
-	data := &CognitoSyncEvent{
+	data := &CognitoIdentitySyncEvent{
 		CreationDate:     dataset.CreationDate,
 		DataStorage:      dataset.DataStorage,
 		DatasetName:      dataset.DatasetName,
@@ -222,7 +222,7 @@ func (a *adapter) sendCognitoEvent(dataset *cognitosync.Dataset, records []*cogn
 	}
 
 	event := cloudevents.NewEvent(cloudevents.VersionV1)
-	event.SetType(v1alpha1.AWSEventType(a.arn.Service, v1alpha1.AWSCognitoGenericEventType))
+	event.SetType(v1alpha1.AWSEventType(a.arn.Service, v1alpha1.AWSCognitoIdentityGenericEventType))
 	event.SetSubject(*dataset.DatasetName)
 	event.SetSource(a.arn.String())
 	event.SetID(*dataset.IdentityId)
diff --git a/pkg/adapter/awscognitosource/events.go b/pkg/adapter/awscognitoidentitysource/events.go
similarity index 86%
rename from pkg/adapter/awscognitosource/events.go
rename to pkg/adapter/awscognitoidentitysource/events.go
index b9b7c37..afc532d 100644
--- a/pkg/adapter/awscognitosource/events.go
+++ b/pkg/adapter/awscognitoidentitysource/events.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	"time"
@@ -22,8 +22,8 @@ import (
 	"github.com/aws/aws-sdk-go/service/cognitosync"
 )
 
-// CognitoSyncEvent represents the payload of a Cognito sync event.
-type CognitoSyncEvent struct {
+// CognitoIdentitySyncEvent represents the payload of a Cognito sync event.
+type CognitoIdentitySyncEvent struct {
 	CreationDate     *time.Time
 	DataStorage      *int64
 	DatasetName      *string
diff --git a/pkg/adapter/awscognitouserpoolsource/adapter.go b/pkg/adapter/awscognitouserpoolsource/adapter.go
index b0cae46..7e6073f 100644
--- a/pkg/adapter/awscognitouserpoolsource/adapter.go
+++ b/pkg/adapter/awscognitouserpoolsource/adapter.go
@@ -83,7 +83,7 @@ func NewAdapter(ctx context.Context, envAcc pkgadapter.EnvConfigAccessor, ceClie
 		ceClient:          ceClient,
 
 		arn:        arn,
-		userPoolID: common.MustParseCognitoUserPoolID(arn.Resource),
+		userPoolID: common.MustParseCognitoUserPoolResource(arn.Resource),
 	}
 }
 
diff --git a/pkg/adapter/awssnssource/adapter.go b/pkg/adapter/awssnssource/adapter.go
index 736e048..dd86663 100644
--- a/pkg/adapter/awssnssource/adapter.go
+++ b/pkg/adapter/awssnssource/adapter.go
@@ -23,7 +23,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"net/url"
-	"os"
+	"sync"
 	"time"
 
 	"go.uber.org/zap"
@@ -48,7 +48,9 @@ import (
 type envConfig struct {
 	pkgadapter.EnvConfig
 
-	ARN string `envconfig:"ARN" required:"true"`
+	ARN                    string `required:"true"`
+	SubscriptionAttributes []byte `split_words:"true"`
+	PublicURL              string `required:"true" envconfig:"PUBLIC_URL"`
 }
 
 // adapter implements the source's adapter.
@@ -58,7 +60,9 @@ type adapter struct {
 	snsClient snsiface.SNSAPI
 	ceClient  cloudevents.Client
 
-	arn arn.ARN
+	arn       arn.ARN
+	subAttrs  map[string]*string
+	publicURL url.URL
 }
 
 // NewEnvConfig returns an accessor for the source's adapter envConfig.
@@ -72,6 +76,9 @@ func NewAdapter(ctx context.Context, envAcc pkgadapter.EnvConfigAccessor, ceClie
 
 	env := envAcc.(*envConfig)
 
+	var subAttrs map[string]*string
+	mustUnmarshalSubscriptionAttributes(env.SubscriptionAttributes, subAttrs)
+
 	arn := common.MustParseARN(env.ARN)
 
 	cfg := session.Must(session.NewSession(aws.NewConfig().
@@ -85,59 +92,135 @@ func NewAdapter(ctx context.Context, envAcc pkgadapter.EnvConfigAccessor, ceClie
 		snsClient: sns.New(cfg),
 		ceClient:  ceClient,
 
-		arn: arn,
+		arn:       arn,
+		subAttrs:  subAttrs,
+		publicURL: mustParsePublicURL(env.PublicURL),
+	}
+}
+
+func mustParsePublicURL(publicURL string) url.URL {
+	url, err := url.Parse(publicURL)
+	if err != nil {
+		panic(err)
+	}
+	if url.String() == "" {
+		panic("empty public URL")
+	}
+
+	return *url
+}
+
+func mustUnmarshalSubscriptionAttributes(data []byte, subAttrs map[string]*string) {
+	if data == nil {
+		return
+	}
+
+	err := json.Unmarshal(data, &subAttrs)
+	if err != nil {
+		panic(err)
 	}
 }
 
 const (
-	port                      = 8081
-	defaultSubscriptionPeriod = 10 * time.Second
+	serverPort                = "8080"
+	serverShutdownGracePeriod = time.Second * 10
+	subscriptionRecheckPeriod = time.Second * 10
 )
 
 // Start implements adapter.Adapter.
 func (a *adapter) Start(stopCh <-chan struct{}) error {
-	// Setup subscription in the background. Will keep us from having chicken/egg between server
-	// being ready to respond and us having the info we need for the subscription request
+	// ctx gets canceled to stop goroutines
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// handle stop signals
 	go func() {
-		for {
-			if err := a.attempSubscription(defaultSubscriptionPeriod); err != nil {
-				a.logger.Error(err)
-			}
-		}
+		<-stopCh
+		a.logger.Info("Shutdown signal received. Terminating")
+		cancel()
 	}()
 
-	// Start server
 	http.HandleFunc("/", a.handleNotification)
 	http.HandleFunc("/health", healthCheckHandler)
-	a.logger.Infof("Serving on port %d", port)
-	return http.ListenAndServe(fmt.Sprintf(":%d", port), nil)
-}
 
-func (a *adapter) attempSubscription(period time.Duration) error {
-	time.Sleep(period)
+	server := &http.Server{Addr: ":" + serverPort}
+	serverErrCh := make(chan error)
+	defer close(serverErrCh)
 
-	topic, err := a.snsClient.CreateTopic(&sns.CreateTopicInput{Name: &a.arn.Resource})
-	if err != nil {
-		return err
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		a.logger.Info("Serving on port " + serverPort)
+		serverErrCh <- server.ListenAndServe()
+		wg.Done()
+	}()
+
+	/* TODO(antoineco): we should delete the subscription when the source
+	   is deleted by can't do it from the adapter because a) it should
+	   scale to zero b) it shouldn't have access to the Kubernetes API to
+	   read the event source object.
+	   Ref. https://github.com/triggermesh/aws-event-sources/issues/157
+	*/
+	wg.Add(1)
+	go func() {
+		a.runSubscriptionReconciler(ctx, subscriptionRecheckPeriod)
+		wg.Done()
+	}()
+
+	var err error
+
+	select {
+	case serverErr := <-serverErrCh:
+		if serverErr != nil {
+			err = fmt.Errorf("failure during runtime of SNS notification handler: %w", serverErr)
+		}
+		cancel()
+
+	case <-ctx.Done():
+		a.logger.Info("Shutting server down")
+
+		shutdownCtx, cancelTimeout := context.WithTimeout(ctx, serverShutdownGracePeriod)
+		defer cancelTimeout()
+		if shutdownErr := server.Shutdown(shutdownCtx); shutdownErr != nil {
+			err = fmt.Errorf("error during server shutdown: %w", shutdownErr)
+		}
+
+		// unblock server goroutine
+		<-serverErrCh
 	}
 
-	sink := os.Getenv("K_SINK")
-	sinkURL, err := url.Parse(sink)
-	if err != nil {
-		return err
+	wg.Wait()
+	return err
+}
+
+func (a *adapter) runSubscriptionReconciler(ctx context.Context, recheckPeriod time.Duration) {
+	ticker := time.NewTicker(recheckPeriod)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			if err := a.reconcileSubscription(); err != nil {
+				a.logger.Errorw("Failed to reconcile Subscription", zap.Error(err))
+			}
+		case <-ctx.Done():
+			a.logger.Info("Shutting subscription reconciler down")
+			return
+		}
 	}
+}
 
-	_, err = a.snsClient.Subscribe(&sns.SubscribeInput{
-		Endpoint: &sink,
-		Protocol: &sinkURL.Scheme,
-		TopicArn: topic.TopicArn,
+func (a *adapter) reconcileSubscription() error {
+	resp, err := a.snsClient.Subscribe(&sns.SubscribeInput{
+		Endpoint:   aws.String(a.publicURL.String()),
+		Protocol:   &a.publicURL.Scheme,
+		TopicArn:   aws.String(a.arn.String()),
+		Attributes: a.subAttrs,
 	})
-	if err != nil {
-		return err
-	}
+	a.logger.Debug("Subscribe responded with: ", resp)
 
-	a.logger.Debug("Finished subscription flow")
-	return nil
+	return err
 }
 
 // handleNotification implements the receive interface for SNS.
@@ -172,54 +255,36 @@ func (a *adapter) handleNotification(rw http.ResponseWriter, r *http.Request) {
 			return
 		}
 
-		a.logger.Debug("Successfully confirmed SNS subscription ", *resp.SubscriptionArn)
+		a.logger.Debug("Successfully confirmed SNS subscription: ", *resp)
 
 	// If the message is a notification, push the event
 	// payload: https://docs.aws.amazon.com/sns/latest/dg/sns-message-and-json-formats.html#http-notification-json
 	case "Notification":
-		eventTime, err := time.Parse(time.RFC3339, data["Timestamp"].(string))
-		if err != nil {
-			a.logger.Errorw("Failed to parse notification timestamp", zap.Error(err))
-			http.Error(rw, fmt.Sprint("Failed to parse notification timestamp: ", err), http.StatusBadRequest)
-			return
-		}
-
-		record := &SNSEventRecord{
-			EventVersion: "1.0",
-			EventSource:  "aws:sns",
-			SNS: SNSEntity{
-				Message:          data["Message"].(string),
-				MessageID:        data["MessageId"].(string),
-				Signature:        data["Signature"].(string),
-				SignatureVersion: data["SignatureVersion"].(string),
-				SigningCertURL:   data["SigningCertURL"].(string),
-				Subject:          data["Subject"].(string),
-				Timestamp:        eventTime,
-				TopicArn:         data["TopicArn"].(string),
-				Type:             data["Type"].(string),
-				UnsubscribeURL:   data["UnsubscribeURL"].(string),
-			},
-		}
-
 		event := cloudevents.NewEvent(cloudevents.VersionV1)
 		event.SetType(v1alpha1.AWSEventType(a.arn.Service, v1alpha1.AWSSNSGenericEventType))
-		event.SetSubject(data["Subject"].(string))
 		event.SetSource(a.arn.String())
 		event.SetID(data["MessageId"].(string))
-		if err := event.SetData(cloudevents.ApplicationJSON, record); err != nil {
+
+		if subjectAttr, ok := data["Subject"]; ok {
+			event.SetSubject(subjectAttr.(string))
+		}
+
+		if err := event.SetData(cloudevents.ApplicationJSON, body); err != nil {
 			a.logger.Errorw("Failed to set event data", zap.Error(err))
 			http.Error(rw, fmt.Sprint("Failed to set event data: ", err), http.StatusInternalServerError)
 			return
 		}
 
 		if result := a.ceClient.Send(context.Background(), event); !cloudevents.IsACK(result) {
-			a.logger.Errorw("Failed to send CloudEvent", "error", result)
+			a.logger.Errorw("Failed to send CloudEvent", zap.Error(result))
 			http.Error(rw, fmt.Sprint("Failed to send CloudEvent: ", result), http.StatusInternalServerError)
 		}
+
+		a.logger.Debug("Successfully sent SNS notification: ", event)
 	}
 }
 
-func healthCheckHandler(w http.ResponseWriter, r *http.Request) {
+func healthCheckHandler(w http.ResponseWriter, _ *http.Request) {
 	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
 	w.WriteHeader(http.StatusOK)
 	fmt.Fprintln(w, "OK")
diff --git a/pkg/adapter/awssnssource/events.go b/pkg/adapter/awssnssource/events.go
deleted file mode 100644
index e071e8c..0000000
--- a/pkg/adapter/awssnssource/events.go
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
-Copyright (c) 2019-2020 TriggerMesh Inc.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-   http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package awssnssource
-
-import (
-	"time"
-)
-
-// SNSEventRecord represents a SNS event.
-type SNSEventRecord struct {
-	EventVersion string    `json:"eventVersion"`
-	EventSource  string    `json:"eventSource"`
-	SNS          SNSEntity `json:"sns"`
-}
-
-// SNSEntity is the data from a SNS notification.
-// see https://docs.aws.amazon.com/sns/latest/dg/sns-message-and-json-formats.html#http-notification-json
-type SNSEntity struct {
-	Message           string                 `json:"message"`
-	MessageID         string                 `json:"messageId"`
-	Signature         string                 `json:"signature"`
-	SignatureVersion  string                 `json:"signatureVersion"`
-	SigningCertURL    string                 `json:"signingCertUrl"`
-	Subject           string                 `json:"subject"`
-	Timestamp         time.Time              `json:"timestamp"`
-	TopicArn          string                 `json:"topicArn"`
-	Type              string                 `json:"type"`
-	UnsubscribeURL    string                 `json:"unsubscribeUrl"`
-	MessageAttributes map[string]interface{} `json:"messageAttributes"`
-}
diff --git a/pkg/adapter/common/common.go b/pkg/adapter/common/common.go
index f976a5c..cb065e7 100644
--- a/pkg/adapter/common/common.go
+++ b/pkg/adapter/common/common.go
@@ -25,8 +25,8 @@ import (
 )
 
 const (
-	expectCognitoResourceFmt         = "identitypool/${IdentityPoolId}"
 	expectCognitoUserPoolResourceFmt = "userpool/${UserPoolId}"
+	expectCognitoIdentityResourceFmt = "identitypool/${IdentityPoolId}"
 	expectDynamoDBResourceFmt        = "table/${TableName}"
 	expectKinesisResourceFmt         = "stream/${StreamName}"
 )
@@ -40,10 +40,10 @@ func MustParseARN(arnStr string) arn.ARN {
 	return arn
 }
 
-// MustParseCognitoResource parses the resource segment of a Cognito ARN and
-// panics in case of error.
-func MustParseCognitoResource(resource string) string /*identityPoolId*/ {
-	elements, err := parseResource(resource, expectCognitoResourceFmt)
+// MustParseCognitoIdentityResource parses the resource segment of a Cognito
+// Identity Pool ARN and panics in case of error.
+func MustParseCognitoIdentityResource(resource string) string /*identityPoolId*/ {
+	elements, err := parseResource(resource, expectCognitoIdentityResourceFmt)
 	if err != nil {
 		panic(err)
 	}
@@ -70,9 +70,9 @@ func MustParseKinesisResource(resource string) string /*stream*/ {
 	return elements[0]
 }
 
-// MustParseCognitoUserPoolID parses the resource segment of a Cognito User Pool
+// MustParseCognitoUserPoolResource parses the resource segment of a Cognito User Pool
 // ARN and panics in case of error.
-func MustParseCognitoUserPoolID(resource string) string {
+func MustParseCognitoUserPoolResource(resource string) string {
 	elements, err := parseResource(resource, expectCognitoUserPoolResourceFmt)
 	if err != nil {
 		panic(err)
diff --git a/pkg/apis/sources/register.go b/pkg/apis/sources/register.go
index 3823149..3c5b104 100644
--- a/pkg/apis/sources/register.go
+++ b/pkg/apis/sources/register.go
@@ -28,14 +28,14 @@ var (
 		Resource: "awscodecommitsources",
 	}
 
-	// AWSCognitoSourceResource respresents an event source for AWS Cognito.
-	AWSCognitoSourceResource = schema.GroupResource{
+	// AWSCognitoIdentitySourceResource respresents an event source for AWS Cognito.
+	AWSCognitoIdentitySourceResource = schema.GroupResource{
 		Group:    GroupName,
-		Resource: "awscognitosources",
+		Resource: "awscognitoidentitysources",
 	}
 
 	// AWSCognitoUserPoolSourceResource respresents an event source for AWS Cognito User Pool.
-	AWSCognitoSourceUserPoolResource = schema.GroupResource{
+	AWSCognitoUserPoolSourceResource = schema.GroupResource{
 		Group:    GroupName,
 		Resource: "awscognitouserpoolsources",
 	}
diff --git a/pkg/apis/sources/v1alpha1/awscognito_defaults.go b/pkg/apis/sources/v1alpha1/awscognitoidentity_defaults.go
similarity index 90%
rename from pkg/apis/sources/v1alpha1/awscognito_defaults.go
rename to pkg/apis/sources/v1alpha1/awscognitoidentity_defaults.go
index 3935d4c..20039df 100644
--- a/pkg/apis/sources/v1alpha1/awscognito_defaults.go
+++ b/pkg/apis/sources/v1alpha1/awscognitoidentity_defaults.go
@@ -19,4 +19,4 @@ package v1alpha1
 import "context"
 
 // SetDefaults implements apis.Defaultable.
-func (s *AWSCognitoSource) SetDefaults(_ context.Context) {}
+func (s *AWSCognitoIdentitySource) SetDefaults(_ context.Context) {}
diff --git a/pkg/apis/sources/v1alpha1/awscognito_lifecycle.go b/pkg/apis/sources/v1alpha1/awscognitoidentity_lifecycle.go
similarity index 65%
rename from pkg/apis/sources/v1alpha1/awscognito_lifecycle.go
rename to pkg/apis/sources/v1alpha1/awscognitoidentity_lifecycle.go
index 24558db..1d7e39d 100644
--- a/pkg/apis/sources/v1alpha1/awscognito_lifecycle.go
+++ b/pkg/apis/sources/v1alpha1/awscognitoidentity_lifecycle.go
@@ -22,38 +22,38 @@ import (
 )
 
 // GetGroupVersionKind implements kmeta.OwnerRefable.
-func (s *AWSCognitoSource) GetGroupVersionKind() schema.GroupVersionKind {
-	return SchemeGroupVersion.WithKind("AWSCognitoSource")
+func (s *AWSCognitoIdentitySource) GetGroupVersionKind() schema.GroupVersionKind {
+	return SchemeGroupVersion.WithKind("AWSCognitoIdentitySource")
 }
 
 // GetUntypedSpec implements apis.HasSpec.
-func (s *AWSCognitoSource) GetUntypedSpec() interface{} {
+func (s *AWSCognitoIdentitySource) GetUntypedSpec() interface{} {
 	return s.Spec
 }
 
 // GetSink implements AWSEventSource.
-func (s *AWSCognitoSource) GetSink() *duckv1.Destination {
+func (s *AWSCognitoIdentitySource) GetSink() *duckv1.Destination {
 	return &s.Spec.Sink
 }
 
 // GetARN implements AWSEventSource.
-func (s *AWSCognitoSource) GetARN() string {
+func (s *AWSCognitoIdentitySource) GetARN() string {
 	return s.Spec.ARN
 }
 
 // GetStatus implements AWSEventSource.
-func (s *AWSCognitoSource) GetStatus() *AWSEventSourceStatus {
+func (s *AWSCognitoIdentitySource) GetStatus() *AWSEventSourceStatus {
 	return &s.Status
 }
 
 // Supported event types
 const (
-	AWSCognitoGenericEventType = "sync_trigger"
+	AWSCognitoIdentityGenericEventType = "sync_trigger"
 )
 
 // GetEventTypes implements AWSEventSource.
-func (s *AWSCognitoSource) GetEventTypes() []string {
+func (s *AWSCognitoIdentitySource) GetEventTypes() []string {
 	return []string{
-		AWSCognitoGenericEventType,
+		AWSCognitoIdentityGenericEventType,
 	}
 }
diff --git a/pkg/apis/sources/v1alpha1/awscognito_types.go b/pkg/apis/sources/v1alpha1/awscognitoidentity_types.go
similarity index 63%
rename from pkg/apis/sources/v1alpha1/awscognito_types.go
rename to pkg/apis/sources/v1alpha1/awscognitoidentity_types.go
index b6792dd..ed3a30b 100644
--- a/pkg/apis/sources/v1alpha1/awscognito_types.go
+++ b/pkg/apis/sources/v1alpha1/awscognitoidentity_types.go
@@ -29,27 +29,27 @@ import (
 // +genreconciler
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
-// AWSCognitoSource is the Schema for the event source.
-type AWSCognitoSource struct {
+// AWSCognitoIdentitySource is the Schema for the event source.
+type AWSCognitoIdentitySource struct {
 	metav1.TypeMeta   `json:",inline"`
 	metav1.ObjectMeta `json:"metadata,omitempty"`
 
-	Spec   AWSCognitoSourceSpec `json:"spec,omitempty"`
-	Status AWSEventSourceStatus `json:"status,omitempty"`
+	Spec   AWSCognitoIdentitySourceSpec `json:"spec,omitempty"`
+	Status AWSEventSourceStatus         `json:"status,omitempty"`
 }
 
 // Check the interfaces the event source should be implementing.
 var (
-	_ runtime.Object     = (*AWSCognitoSource)(nil)
-	_ kmeta.OwnerRefable = (*AWSCognitoSource)(nil)
-	_ apis.Validatable   = (*AWSCognitoSource)(nil)
-	_ apis.Defaultable   = (*AWSCognitoSource)(nil)
-	_ apis.HasSpec       = (*AWSCognitoSource)(nil)
-	_ AWSEventSource     = (*AWSCognitoSource)(nil)
+	_ runtime.Object     = (*AWSCognitoIdentitySource)(nil)
+	_ kmeta.OwnerRefable = (*AWSCognitoIdentitySource)(nil)
+	_ apis.Validatable   = (*AWSCognitoIdentitySource)(nil)
+	_ apis.Defaultable   = (*AWSCognitoIdentitySource)(nil)
+	_ apis.HasSpec       = (*AWSCognitoIdentitySource)(nil)
+	_ AWSEventSource     = (*AWSCognitoIdentitySource)(nil)
 )
 
-// AWSCognitoSourceSpec defines the desired state of the event source.
-type AWSCognitoSourceSpec struct {
+// AWSCognitoIdentitySourceSpec defines the desired state of the event source.
+type AWSCognitoIdentitySourceSpec struct {
 	duckv1.SourceSpec `json:",inline"`
 
 	// Identity Pool ARN
@@ -62,9 +62,9 @@ type AWSCognitoSourceSpec struct {
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
-// AWSCognitoSourceList contains a list of event sources.
-type AWSCognitoSourceList struct {
+// AWSCognitoIdentitySourceList contains a list of event sources.
+type AWSCognitoIdentitySourceList struct {
 	metav1.TypeMeta `json:",inline"`
 	metav1.ListMeta `json:"metadata,omitempty"`
-	Items           []AWSCognitoSource `json:"items"`
+	Items           []AWSCognitoIdentitySource `json:"items"`
 }
diff --git a/pkg/apis/sources/v1alpha1/awscognito_validation.go b/pkg/apis/sources/v1alpha1/awscognitoidentity_validation.go
similarity index 87%
rename from pkg/apis/sources/v1alpha1/awscognito_validation.go
rename to pkg/apis/sources/v1alpha1/awscognitoidentity_validation.go
index 849b24b..75c132b 100644
--- a/pkg/apis/sources/v1alpha1/awscognito_validation.go
+++ b/pkg/apis/sources/v1alpha1/awscognitoidentity_validation.go
@@ -23,4 +23,4 @@ import (
 )
 
 // Validate implements apis.Validatable.
-func (s *AWSCognitoSource) Validate(_ context.Context) *apis.FieldError { return nil }
+func (s *AWSCognitoIdentitySource) Validate(_ context.Context) *apis.FieldError { return nil }
diff --git a/pkg/apis/sources/v1alpha1/awscognitouserpool_lifecycle.go b/pkg/apis/sources/v1alpha1/awscognitouserpool_lifecycle.go
index 2c7b0b2..cf2e540 100644
--- a/pkg/apis/sources/v1alpha1/awscognitouserpool_lifecycle.go
+++ b/pkg/apis/sources/v1alpha1/awscognitouserpool_lifecycle.go
@@ -48,12 +48,12 @@ func (s *AWSCognitoUserPoolSource) GetStatus() *AWSEventSourceStatus {
 
 // Supported event types
 const (
-	AWSCognitoUserPoolGenericEventType = "sync_trigger"
+	AWSCognitoGenericEventType = "sync_trigger"
 )
 
 // GetEventTypes implements AWSEventSource.
 func (s *AWSCognitoUserPoolSource) GetEventTypes() []string {
 	return []string{
-		AWSCognitoUserPoolGenericEventType,
+		AWSCognitoGenericEventType,
 	}
 }
diff --git a/pkg/apis/sources/v1alpha1/awssns_types.go b/pkg/apis/sources/v1alpha1/awssns_types.go
index 46b8f22..838810b 100644
--- a/pkg/apis/sources/v1alpha1/awssns_types.go
+++ b/pkg/apis/sources/v1alpha1/awssns_types.go
@@ -56,6 +56,13 @@ type AWSSNSSourceSpec struct {
 	// https://docs.aws.amazon.com/IAM/latest/UserGuide/list_amazonsns.html#amazonsns-resources-for-iam-policies
 	ARN string `json:"arn"`
 
+	// Attributes to set on the Subscription.
+	// For a list of supported subscription attributes, please refer to the following resources:
+	//  * https://docs.aws.amazon.com/sns/latest/api/API_SetSubscriptionAttributes.html
+	//  * https://docs.aws.amazon.com/sns/latest/dg/sns-how-it-works.html
+	// +optional
+	SubscriptionAttributes map[string]*string `json:"subscriptionAttributes,omitempty"`
+
 	// Credentials to interact with the AWS SNS API.
 	Credentials AWSSecurityCredentials `json:"credentials"`
 }
diff --git a/pkg/apis/sources/v1alpha1/common_lifecycle.go b/pkg/apis/sources/v1alpha1/common_lifecycle.go
index 67b937e..5781afa 100644
--- a/pkg/apis/sources/v1alpha1/common_lifecycle.go
+++ b/pkg/apis/sources/v1alpha1/common_lifecycle.go
@@ -21,6 +21,8 @@ import (
 
 	"knative.dev/eventing/pkg/apis/duck"
 	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	servingv1 "knative.dev/serving/pkg/apis/serving/v1"
 )
 
 // AWSEventType returns an event type in a format suitable for usage as a
@@ -59,9 +61,22 @@ func (s *AWSEventSourceStatus) MarkNoSink() {
 		ReasonSinkNotFound, "The sink does not exist or its URI is not set")
 }
 
-// PropagateAvailability uses the readiness of the provided Deployment to
-// determine whether the Deployed condition should be marked as true or false.
-func (s *AWSEventSourceStatus) PropagateAvailability(d *appsv1.Deployment) {
+// PropagateAvailability uses the readiness of the provided Deployment or
+// Service to determine whether the Deployed condition should be marked as True
+// or False.
+func (s *AWSEventSourceStatus) PropagateAvailability(obj interface{}) {
+	switch o := obj.(type) {
+	case *appsv1.Deployment:
+		s.propagateDeploymentAvailability(o)
+	case *servingv1.Service:
+		s.propagateServiceAvailability(o)
+	}
+}
+
+func (s *AWSEventSourceStatus) propagateDeploymentAvailability(d *appsv1.Deployment) {
+	// Deployments are not addressable
+	s.Address = nil
+
 	if d == nil {
 		awsEventSourceConditionSet.Manage(s).MarkUnknown(ConditionDeployed, ReasonUnavailable,
 			"The status of the adapter Deployment can not be determined")
@@ -83,3 +98,29 @@ func (s *AWSEventSourceStatus) PropagateAvailability(d *appsv1.Deployment) {
 
 	awsEventSourceConditionSet.Manage(s).MarkFalse(ConditionDeployed, ReasonUnavailable, msg)
 }
+
+func (s *AWSEventSourceStatus) propagateServiceAvailability(ksvc *servingv1.Service) {
+	if ksvc == nil {
+		awsEventSourceConditionSet.Manage(s).MarkUnknown(ConditionDeployed, ReasonUnavailable,
+			"The status of the adapter Service can not be determined")
+		return
+	}
+
+	if s.Address == nil {
+		s.Address = &duckv1.Addressable{}
+	}
+	s.Address.URL = ksvc.Status.URL
+
+	if ksvc.Status.IsReady() {
+		awsEventSourceConditionSet.Manage(s).MarkTrue(ConditionDeployed)
+		return
+	}
+
+	msg := "The adapter Service is unavailable"
+	readyCond := ksvc.Status.GetCondition(servingv1.ServiceConditionReady)
+	if readyCond != nil && readyCond.Message != "" {
+		msg += ": " + readyCond.Message
+	}
+
+	awsEventSourceConditionSet.Manage(s).MarkFalse(ConditionDeployed, ReasonUnavailable, msg)
+}
diff --git a/pkg/apis/sources/v1alpha1/common_types.go b/pkg/apis/sources/v1alpha1/common_types.go
index 9443bf1..7401876 100644
--- a/pkg/apis/sources/v1alpha1/common_types.go
+++ b/pkg/apis/sources/v1alpha1/common_types.go
@@ -23,7 +23,8 @@ import (
 
 // AWSEventSourceStatus defines the observed state of an event source.
 type AWSEventSourceStatus struct {
-	duckv1.SourceStatus `json:",inline"`
+	duckv1.SourceStatus  `json:",inline"`
+	duckv1.AddressStatus `json:",inline"`
 }
 
 // AWSSecurityCredentials represents a set of AWS security credentials.
diff --git a/pkg/apis/sources/v1alpha1/deepcopy_generated.go b/pkg/apis/sources/v1alpha1/deepcopy_generated.go
index 0656071..d594903 100644
--- a/pkg/apis/sources/v1alpha1/deepcopy_generated.go
+++ b/pkg/apis/sources/v1alpha1/deepcopy_generated.go
@@ -110,7 +110,7 @@ func (in *AWSCodeCommitSourceSpec) DeepCopy() *AWSCodeCommitSourceSpec {
 }
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *AWSCognitoSource) DeepCopyInto(out *AWSCognitoSource) {
+func (in *AWSCognitoIdentitySource) DeepCopyInto(out *AWSCognitoIdentitySource) {
 	*out = *in
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
@@ -119,18 +119,18 @@ func (in *AWSCognitoSource) DeepCopyInto(out *AWSCognitoSource) {
 	return
 }
 
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AWSCognitoSource.
-func (in *AWSCognitoSource) DeepCopy() *AWSCognitoSource {
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AWSCognitoIdentitySource.
+func (in *AWSCognitoIdentitySource) DeepCopy() *AWSCognitoIdentitySource {
 	if in == nil {
 		return nil
 	}
-	out := new(AWSCognitoSource)
+	out := new(AWSCognitoIdentitySource)
 	in.DeepCopyInto(out)
 	return out
 }
 
 // DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
-func (in *AWSCognitoSource) DeepCopyObject() runtime.Object {
+func (in *AWSCognitoIdentitySource) DeepCopyObject() runtime.Object {
 	if c := in.DeepCopy(); c != nil {
 		return c
 	}
@@ -138,13 +138,13 @@ func (in *AWSCognitoSource) DeepCopyObject() runtime.Object {
 }
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *AWSCognitoSourceList) DeepCopyInto(out *AWSCognitoSourceList) {
+func (in *AWSCognitoIdentitySourceList) DeepCopyInto(out *AWSCognitoIdentitySourceList) {
 	*out = *in
 	out.TypeMeta = in.TypeMeta
 	in.ListMeta.DeepCopyInto(&out.ListMeta)
 	if in.Items != nil {
 		in, out := &in.Items, &out.Items
-		*out = make([]AWSCognitoSource, len(*in))
+		*out = make([]AWSCognitoIdentitySource, len(*in))
 		for i := range *in {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
@@ -152,18 +152,18 @@ func (in *AWSCognitoSourceList) DeepCopyInto(out *AWSCognitoSourceList) {
 	return
 }
 
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AWSCognitoSourceList.
-func (in *AWSCognitoSourceList) DeepCopy() *AWSCognitoSourceList {
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AWSCognitoIdentitySourceList.
+func (in *AWSCognitoIdentitySourceList) DeepCopy() *AWSCognitoIdentitySourceList {
 	if in == nil {
 		return nil
 	}
-	out := new(AWSCognitoSourceList)
+	out := new(AWSCognitoIdentitySourceList)
 	in.DeepCopyInto(out)
 	return out
 }
 
 // DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
-func (in *AWSCognitoSourceList) DeepCopyObject() runtime.Object {
+func (in *AWSCognitoIdentitySourceList) DeepCopyObject() runtime.Object {
 	if c := in.DeepCopy(); c != nil {
 		return c
 	}
@@ -171,19 +171,19 @@ func (in *AWSCognitoSourceList) DeepCopyObject() runtime.Object {
 }
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *AWSCognitoSourceSpec) DeepCopyInto(out *AWSCognitoSourceSpec) {
+func (in *AWSCognitoIdentitySourceSpec) DeepCopyInto(out *AWSCognitoIdentitySourceSpec) {
 	*out = *in
 	in.SourceSpec.DeepCopyInto(&out.SourceSpec)
 	in.Credentials.DeepCopyInto(&out.Credentials)
 	return
 }
 
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AWSCognitoSourceSpec.
-func (in *AWSCognitoSourceSpec) DeepCopy() *AWSCognitoSourceSpec {
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AWSCognitoIdentitySourceSpec.
+func (in *AWSCognitoIdentitySourceSpec) DeepCopy() *AWSCognitoIdentitySourceSpec {
 	if in == nil {
 		return nil
 	}
-	out := new(AWSCognitoSourceSpec)
+	out := new(AWSCognitoIdentitySourceSpec)
 	in.DeepCopyInto(out)
 	return out
 }
diff --git a/pkg/apis/sources/v1alpha1/register.go b/pkg/apis/sources/v1alpha1/register.go
index 2152ea3..9f157c4 100644
--- a/pkg/apis/sources/v1alpha1/register.go
+++ b/pkg/apis/sources/v1alpha1/register.go
@@ -37,7 +37,7 @@ var (
 func addKnownTypes(scheme *runtime.Scheme) error {
 	scheme.AddKnownTypes(SchemeGroupVersion,
 		&AWSCodeCommitSource{}, &AWSCodeCommitSourceList{},
-		&AWSCognitoSource{}, &AWSCognitoSourceList{},
+		&AWSCognitoIdentitySource{}, &AWSCognitoUserPoolSourceList{},
 		&AWSCognitoUserPoolSource{}, &AWSCognitoUserPoolSourceList{},
 		&AWSDynamoDBSource{}, &AWSDynamoDBSourceList{},
 		&AWSIoTSource{}, &AWSIoTSourceList{},
diff --git a/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/awscognitoidentitysource.go b/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/awscognitoidentitysource.go
new file mode 100644
index 0000000..211f532
--- /dev/null
+++ b/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/awscognitoidentitysource.go
@@ -0,0 +1,191 @@
+/*
+Copyright (c) 2020 TriggerMesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	"time"
+
+	v1alpha1 "github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
+	scheme "github.com/triggermesh/aws-event-sources/pkg/client/generated/clientset/internalclientset/scheme"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	rest "k8s.io/client-go/rest"
+)
+
+// AWSCognitoIdentitySourcesGetter has a method to return a AWSCognitoIdentitySourceInterface.
+// A group's client should implement this interface.
+type AWSCognitoIdentitySourcesGetter interface {
+	AWSCognitoIdentitySources(namespace string) AWSCognitoIdentitySourceInterface
+}
+
+// AWSCognitoIdentitySourceInterface has methods to work with AWSCognitoIdentitySource resources.
+type AWSCognitoIdentitySourceInterface interface {
+	Create(*v1alpha1.AWSCognitoIdentitySource) (*v1alpha1.AWSCognitoIdentitySource, error)
+	Update(*v1alpha1.AWSCognitoIdentitySource) (*v1alpha1.AWSCognitoIdentitySource, error)
+	UpdateStatus(*v1alpha1.AWSCognitoIdentitySource) (*v1alpha1.AWSCognitoIdentitySource, error)
+	Delete(name string, options *v1.DeleteOptions) error
+	DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error
+	Get(name string, options v1.GetOptions) (*v1alpha1.AWSCognitoIdentitySource, error)
+	List(opts v1.ListOptions) (*v1alpha1.AWSCognitoIdentitySourceList, error)
+	Watch(opts v1.ListOptions) (watch.Interface, error)
+	Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1alpha1.AWSCognitoIdentitySource, err error)
+	AWSCognitoIdentitySourceExpansion
+}
+
+// aWSCognitoIdentitySources implements AWSCognitoIdentitySourceInterface
+type aWSCognitoIdentitySources struct {
+	client rest.Interface
+	ns     string
+}
+
+// newAWSCognitoIdentitySources returns a AWSCognitoIdentitySources
+func newAWSCognitoIdentitySources(c *SourcesV1alpha1Client, namespace string) *aWSCognitoIdentitySources {
+	return &aWSCognitoIdentitySources{
+		client: c.RESTClient(),
+		ns:     namespace,
+	}
+}
+
+// Get takes name of the aWSCognitoIdentitySource, and returns the corresponding aWSCognitoIdentitySource object, and an error if there is any.
+func (c *aWSCognitoIdentitySources) Get(name string, options v1.GetOptions) (result *v1alpha1.AWSCognitoIdentitySource, err error) {
+	result = &v1alpha1.AWSCognitoIdentitySource{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("awscognitoidentitysources").
+		Name(name).
+		VersionedParams(&options, scheme.ParameterCodec).
+		Do().
+		Into(result)
+	return
+}
+
+// List takes label and field selectors, and returns the list of AWSCognitoIdentitySources that match those selectors.
+func (c *aWSCognitoIdentitySources) List(opts v1.ListOptions) (result *v1alpha1.AWSCognitoIdentitySourceList, err error) {
+	var timeout time.Duration
+	if opts.TimeoutSeconds != nil {
+		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
+	}
+	result = &v1alpha1.AWSCognitoIdentitySourceList{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("awscognitoidentitysources").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Do().
+		Into(result)
+	return
+}
+
+// Watch returns a watch.Interface that watches the requested aWSCognitoIdentitySources.
+func (c *aWSCognitoIdentitySources) Watch(opts v1.ListOptions) (watch.Interface, error) {
+	var timeout time.Duration
+	if opts.TimeoutSeconds != nil {
+		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
+	}
+	opts.Watch = true
+	return c.client.Get().
+		Namespace(c.ns).
+		Resource("awscognitoidentitysources").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Watch()
+}
+
+// Create takes the representation of a aWSCognitoIdentitySource and creates it.  Returns the server's representation of the aWSCognitoIdentitySource, and an error, if there is any.
+func (c *aWSCognitoIdentitySources) Create(aWSCognitoIdentitySource *v1alpha1.AWSCognitoIdentitySource) (result *v1alpha1.AWSCognitoIdentitySource, err error) {
+	result = &v1alpha1.AWSCognitoIdentitySource{}
+	err = c.client.Post().
+		Namespace(c.ns).
+		Resource("awscognitoidentitysources").
+		Body(aWSCognitoIdentitySource).
+		Do().
+		Into(result)
+	return
+}
+
+// Update takes the representation of a aWSCognitoIdentitySource and updates it. Returns the server's representation of the aWSCognitoIdentitySource, and an error, if there is any.
+func (c *aWSCognitoIdentitySources) Update(aWSCognitoIdentitySource *v1alpha1.AWSCognitoIdentitySource) (result *v1alpha1.AWSCognitoIdentitySource, err error) {
+	result = &v1alpha1.AWSCognitoIdentitySource{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("awscognitoidentitysources").
+		Name(aWSCognitoIdentitySource.Name).
+		Body(aWSCognitoIdentitySource).
+		Do().
+		Into(result)
+	return
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+
+func (c *aWSCognitoIdentitySources) UpdateStatus(aWSCognitoIdentitySource *v1alpha1.AWSCognitoIdentitySource) (result *v1alpha1.AWSCognitoIdentitySource, err error) {
+	result = &v1alpha1.AWSCognitoIdentitySource{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("awscognitoidentitysources").
+		Name(aWSCognitoIdentitySource.Name).
+		SubResource("status").
+		Body(aWSCognitoIdentitySource).
+		Do().
+		Into(result)
+	return
+}
+
+// Delete takes name of the aWSCognitoIdentitySource and deletes it. Returns an error if one occurs.
+func (c *aWSCognitoIdentitySources) Delete(name string, options *v1.DeleteOptions) error {
+	return c.client.Delete().
+		Namespace(c.ns).
+		Resource("awscognitoidentitysources").
+		Name(name).
+		Body(options).
+		Do().
+		Error()
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *aWSCognitoIdentitySources) DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error {
+	var timeout time.Duration
+	if listOptions.TimeoutSeconds != nil {
+		timeout = time.Duration(*listOptions.TimeoutSeconds) * time.Second
+	}
+	return c.client.Delete().
+		Namespace(c.ns).
+		Resource("awscognitoidentitysources").
+		VersionedParams(&listOptions, scheme.ParameterCodec).
+		Timeout(timeout).
+		Body(options).
+		Do().
+		Error()
+}
+
+// Patch applies the patch and returns the patched aWSCognitoIdentitySource.
+func (c *aWSCognitoIdentitySources) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1alpha1.AWSCognitoIdentitySource, err error) {
+	result = &v1alpha1.AWSCognitoIdentitySource{}
+	err = c.client.Patch(pt).
+		Namespace(c.ns).
+		Resource("awscognitoidentitysources").
+		SubResource(subresources...).
+		Name(name).
+		Body(data).
+		Do().
+		Into(result)
+	return
+}
diff --git a/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/awscognitosource.go b/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/awscognitosource.go
deleted file mode 100644
index 122cf44..0000000
--- a/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/awscognitosource.go
+++ /dev/null
@@ -1,191 +0,0 @@
-/*
-Copyright (c) 2020 TriggerMesh Inc.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-   http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-// Code generated by client-gen. DO NOT EDIT.
-
-package v1alpha1
-
-import (
-	"time"
-
-	v1alpha1 "github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
-	scheme "github.com/triggermesh/aws-event-sources/pkg/client/generated/clientset/internalclientset/scheme"
-	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	types "k8s.io/apimachinery/pkg/types"
-	watch "k8s.io/apimachinery/pkg/watch"
-	rest "k8s.io/client-go/rest"
-)
-
-// AWSCognitoSourcesGetter has a method to return a AWSCognitoSourceInterface.
-// A group's client should implement this interface.
-type AWSCognitoSourcesGetter interface {
-	AWSCognitoSources(namespace string) AWSCognitoSourceInterface
-}
-
-// AWSCognitoSourceInterface has methods to work with AWSCognitoSource resources.
-type AWSCognitoSourceInterface interface {
-	Create(*v1alpha1.AWSCognitoSource) (*v1alpha1.AWSCognitoSource, error)
-	Update(*v1alpha1.AWSCognitoSource) (*v1alpha1.AWSCognitoSource, error)
-	UpdateStatus(*v1alpha1.AWSCognitoSource) (*v1alpha1.AWSCognitoSource, error)
-	Delete(name string, options *v1.DeleteOptions) error
-	DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error
-	Get(name string, options v1.GetOptions) (*v1alpha1.AWSCognitoSource, error)
-	List(opts v1.ListOptions) (*v1alpha1.AWSCognitoSourceList, error)
-	Watch(opts v1.ListOptions) (watch.Interface, error)
-	Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1alpha1.AWSCognitoSource, err error)
-	AWSCognitoSourceExpansion
-}
-
-// aWSCognitoSources implements AWSCognitoSourceInterface
-type aWSCognitoSources struct {
-	client rest.Interface
-	ns     string
-}
-
-// newAWSCognitoSources returns a AWSCognitoSources
-func newAWSCognitoSources(c *SourcesV1alpha1Client, namespace string) *aWSCognitoSources {
-	return &aWSCognitoSources{
-		client: c.RESTClient(),
-		ns:     namespace,
-	}
-}
-
-// Get takes name of the aWSCognitoSource, and returns the corresponding aWSCognitoSource object, and an error if there is any.
-func (c *aWSCognitoSources) Get(name string, options v1.GetOptions) (result *v1alpha1.AWSCognitoSource, err error) {
-	result = &v1alpha1.AWSCognitoSource{}
-	err = c.client.Get().
-		Namespace(c.ns).
-		Resource("awscognitosources").
-		Name(name).
-		VersionedParams(&options, scheme.ParameterCodec).
-		Do().
-		Into(result)
-	return
-}
-
-// List takes label and field selectors, and returns the list of AWSCognitoSources that match those selectors.
-func (c *aWSCognitoSources) List(opts v1.ListOptions) (result *v1alpha1.AWSCognitoSourceList, err error) {
-	var timeout time.Duration
-	if opts.TimeoutSeconds != nil {
-		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
-	}
-	result = &v1alpha1.AWSCognitoSourceList{}
-	err = c.client.Get().
-		Namespace(c.ns).
-		Resource("awscognitosources").
-		VersionedParams(&opts, scheme.ParameterCodec).
-		Timeout(timeout).
-		Do().
-		Into(result)
-	return
-}
-
-// Watch returns a watch.Interface that watches the requested aWSCognitoSources.
-func (c *aWSCognitoSources) Watch(opts v1.ListOptions) (watch.Interface, error) {
-	var timeout time.Duration
-	if opts.TimeoutSeconds != nil {
-		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
-	}
-	opts.Watch = true
-	return c.client.Get().
-		Namespace(c.ns).
-		Resource("awscognitosources").
-		VersionedParams(&opts, scheme.ParameterCodec).
-		Timeout(timeout).
-		Watch()
-}
-
-// Create takes the representation of a aWSCognitoSource and creates it.  Returns the server's representation of the aWSCognitoSource, and an error, if there is any.
-func (c *aWSCognitoSources) Create(aWSCognitoSource *v1alpha1.AWSCognitoSource) (result *v1alpha1.AWSCognitoSource, err error) {
-	result = &v1alpha1.AWSCognitoSource{}
-	err = c.client.Post().
-		Namespace(c.ns).
-		Resource("awscognitosources").
-		Body(aWSCognitoSource).
-		Do().
-		Into(result)
-	return
-}
-
-// Update takes the representation of a aWSCognitoSource and updates it. Returns the server's representation of the aWSCognitoSource, and an error, if there is any.
-func (c *aWSCognitoSources) Update(aWSCognitoSource *v1alpha1.AWSCognitoSource) (result *v1alpha1.AWSCognitoSource, err error) {
-	result = &v1alpha1.AWSCognitoSource{}
-	err = c.client.Put().
-		Namespace(c.ns).
-		Resource("awscognitosources").
-		Name(aWSCognitoSource.Name).
-		Body(aWSCognitoSource).
-		Do().
-		Into(result)
-	return
-}
-
-// UpdateStatus was generated because the type contains a Status member.
-// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
-
-func (c *aWSCognitoSources) UpdateStatus(aWSCognitoSource *v1alpha1.AWSCognitoSource) (result *v1alpha1.AWSCognitoSource, err error) {
-	result = &v1alpha1.AWSCognitoSource{}
-	err = c.client.Put().
-		Namespace(c.ns).
-		Resource("awscognitosources").
-		Name(aWSCognitoSource.Name).
-		SubResource("status").
-		Body(aWSCognitoSource).
-		Do().
-		Into(result)
-	return
-}
-
-// Delete takes name of the aWSCognitoSource and deletes it. Returns an error if one occurs.
-func (c *aWSCognitoSources) Delete(name string, options *v1.DeleteOptions) error {
-	return c.client.Delete().
-		Namespace(c.ns).
-		Resource("awscognitosources").
-		Name(name).
-		Body(options).
-		Do().
-		Error()
-}
-
-// DeleteCollection deletes a collection of objects.
-func (c *aWSCognitoSources) DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error {
-	var timeout time.Duration
-	if listOptions.TimeoutSeconds != nil {
-		timeout = time.Duration(*listOptions.TimeoutSeconds) * time.Second
-	}
-	return c.client.Delete().
-		Namespace(c.ns).
-		Resource("awscognitosources").
-		VersionedParams(&listOptions, scheme.ParameterCodec).
-		Timeout(timeout).
-		Body(options).
-		Do().
-		Error()
-}
-
-// Patch applies the patch and returns the patched aWSCognitoSource.
-func (c *aWSCognitoSources) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1alpha1.AWSCognitoSource, err error) {
-	result = &v1alpha1.AWSCognitoSource{}
-	err = c.client.Patch(pt).
-		Namespace(c.ns).
-		Resource("awscognitosources").
-		SubResource(subresources...).
-		Name(name).
-		Body(data).
-		Do().
-		Into(result)
-	return
-}
diff --git a/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/fake/fake_awscognitoidentitysource.go b/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/fake/fake_awscognitoidentitysource.go
new file mode 100644
index 0000000..a919e87
--- /dev/null
+++ b/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/fake/fake_awscognitoidentitysource.go
@@ -0,0 +1,140 @@
+/*
+Copyright (c) 2020 TriggerMesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	v1alpha1 "github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+)
+
+// FakeAWSCognitoIdentitySources implements AWSCognitoIdentitySourceInterface
+type FakeAWSCognitoIdentitySources struct {
+	Fake *FakeSourcesV1alpha1
+	ns   string
+}
+
+var awscognitoidentitysourcesResource = schema.GroupVersionResource{Group: "sources.triggermesh.io", Version: "v1alpha1", Resource: "awscognitoidentitysources"}
+
+var awscognitoidentitysourcesKind = schema.GroupVersionKind{Group: "sources.triggermesh.io", Version: "v1alpha1", Kind: "AWSCognitoIdentitySource"}
+
+// Get takes name of the aWSCognitoIdentitySource, and returns the corresponding aWSCognitoIdentitySource object, and an error if there is any.
+func (c *FakeAWSCognitoIdentitySources) Get(name string, options v1.GetOptions) (result *v1alpha1.AWSCognitoIdentitySource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(awscognitoidentitysourcesResource, c.ns, name), &v1alpha1.AWSCognitoIdentitySource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.AWSCognitoIdentitySource), err
+}
+
+// List takes label and field selectors, and returns the list of AWSCognitoIdentitySources that match those selectors.
+func (c *FakeAWSCognitoIdentitySources) List(opts v1.ListOptions) (result *v1alpha1.AWSCognitoIdentitySourceList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(awscognitoidentitysourcesResource, awscognitoidentitysourcesKind, c.ns, opts), &v1alpha1.AWSCognitoIdentitySourceList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &v1alpha1.AWSCognitoIdentitySourceList{ListMeta: obj.(*v1alpha1.AWSCognitoIdentitySourceList).ListMeta}
+	for _, item := range obj.(*v1alpha1.AWSCognitoIdentitySourceList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested aWSCognitoIdentitySources.
+func (c *FakeAWSCognitoIdentitySources) Watch(opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(awscognitoidentitysourcesResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a aWSCognitoIdentitySource and creates it.  Returns the server's representation of the aWSCognitoIdentitySource, and an error, if there is any.
+func (c *FakeAWSCognitoIdentitySources) Create(aWSCognitoIdentitySource *v1alpha1.AWSCognitoIdentitySource) (result *v1alpha1.AWSCognitoIdentitySource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(awscognitoidentitysourcesResource, c.ns, aWSCognitoIdentitySource), &v1alpha1.AWSCognitoIdentitySource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.AWSCognitoIdentitySource), err
+}
+
+// Update takes the representation of a aWSCognitoIdentitySource and updates it. Returns the server's representation of the aWSCognitoIdentitySource, and an error, if there is any.
+func (c *FakeAWSCognitoIdentitySources) Update(aWSCognitoIdentitySource *v1alpha1.AWSCognitoIdentitySource) (result *v1alpha1.AWSCognitoIdentitySource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(awscognitoidentitysourcesResource, c.ns, aWSCognitoIdentitySource), &v1alpha1.AWSCognitoIdentitySource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.AWSCognitoIdentitySource), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeAWSCognitoIdentitySources) UpdateStatus(aWSCognitoIdentitySource *v1alpha1.AWSCognitoIdentitySource) (*v1alpha1.AWSCognitoIdentitySource, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(awscognitoidentitysourcesResource, "status", c.ns, aWSCognitoIdentitySource), &v1alpha1.AWSCognitoIdentitySource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.AWSCognitoIdentitySource), err
+}
+
+// Delete takes name of the aWSCognitoIdentitySource and deletes it. Returns an error if one occurs.
+func (c *FakeAWSCognitoIdentitySources) Delete(name string, options *v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(awscognitoidentitysourcesResource, c.ns, name), &v1alpha1.AWSCognitoIdentitySource{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeAWSCognitoIdentitySources) DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(awscognitoidentitysourcesResource, c.ns, listOptions)
+
+	_, err := c.Fake.Invokes(action, &v1alpha1.AWSCognitoIdentitySourceList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched aWSCognitoIdentitySource.
+func (c *FakeAWSCognitoIdentitySources) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1alpha1.AWSCognitoIdentitySource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(awscognitoidentitysourcesResource, c.ns, name, pt, data, subresources...), &v1alpha1.AWSCognitoIdentitySource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.AWSCognitoIdentitySource), err
+}
diff --git a/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/fake/fake_awscognitosource.go b/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/fake/fake_awscognitosource.go
deleted file mode 100644
index 317532b..0000000
--- a/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/fake/fake_awscognitosource.go
+++ /dev/null
@@ -1,140 +0,0 @@
-/*
-Copyright (c) 2020 TriggerMesh Inc.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-   http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-// Code generated by client-gen. DO NOT EDIT.
-
-package fake
-
-import (
-	v1alpha1 "github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
-	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	labels "k8s.io/apimachinery/pkg/labels"
-	schema "k8s.io/apimachinery/pkg/runtime/schema"
-	types "k8s.io/apimachinery/pkg/types"
-	watch "k8s.io/apimachinery/pkg/watch"
-	testing "k8s.io/client-go/testing"
-)
-
-// FakeAWSCognitoSources implements AWSCognitoSourceInterface
-type FakeAWSCognitoSources struct {
-	Fake *FakeSourcesV1alpha1
-	ns   string
-}
-
-var awscognitosourcesResource = schema.GroupVersionResource{Group: "sources.triggermesh.io", Version: "v1alpha1", Resource: "awscognitosources"}
-
-var awscognitosourcesKind = schema.GroupVersionKind{Group: "sources.triggermesh.io", Version: "v1alpha1", Kind: "AWSCognitoSource"}
-
-// Get takes name of the aWSCognitoSource, and returns the corresponding aWSCognitoSource object, and an error if there is any.
-func (c *FakeAWSCognitoSources) Get(name string, options v1.GetOptions) (result *v1alpha1.AWSCognitoSource, err error) {
-	obj, err := c.Fake.
-		Invokes(testing.NewGetAction(awscognitosourcesResource, c.ns, name), &v1alpha1.AWSCognitoSource{})
-
-	if obj == nil {
-		return nil, err
-	}
-	return obj.(*v1alpha1.AWSCognitoSource), err
-}
-
-// List takes label and field selectors, and returns the list of AWSCognitoSources that match those selectors.
-func (c *FakeAWSCognitoSources) List(opts v1.ListOptions) (result *v1alpha1.AWSCognitoSourceList, err error) {
-	obj, err := c.Fake.
-		Invokes(testing.NewListAction(awscognitosourcesResource, awscognitosourcesKind, c.ns, opts), &v1alpha1.AWSCognitoSourceList{})
-
-	if obj == nil {
-		return nil, err
-	}
-
-	label, _, _ := testing.ExtractFromListOptions(opts)
-	if label == nil {
-		label = labels.Everything()
-	}
-	list := &v1alpha1.AWSCognitoSourceList{ListMeta: obj.(*v1alpha1.AWSCognitoSourceList).ListMeta}
-	for _, item := range obj.(*v1alpha1.AWSCognitoSourceList).Items {
-		if label.Matches(labels.Set(item.Labels)) {
-			list.Items = append(list.Items, item)
-		}
-	}
-	return list, err
-}
-
-// Watch returns a watch.Interface that watches the requested aWSCognitoSources.
-func (c *FakeAWSCognitoSources) Watch(opts v1.ListOptions) (watch.Interface, error) {
-	return c.Fake.
-		InvokesWatch(testing.NewWatchAction(awscognitosourcesResource, c.ns, opts))
-
-}
-
-// Create takes the representation of a aWSCognitoSource and creates it.  Returns the server's representation of the aWSCognitoSource, and an error, if there is any.
-func (c *FakeAWSCognitoSources) Create(aWSCognitoSource *v1alpha1.AWSCognitoSource) (result *v1alpha1.AWSCognitoSource, err error) {
-	obj, err := c.Fake.
-		Invokes(testing.NewCreateAction(awscognitosourcesResource, c.ns, aWSCognitoSource), &v1alpha1.AWSCognitoSource{})
-
-	if obj == nil {
-		return nil, err
-	}
-	return obj.(*v1alpha1.AWSCognitoSource), err
-}
-
-// Update takes the representation of a aWSCognitoSource and updates it. Returns the server's representation of the aWSCognitoSource, and an error, if there is any.
-func (c *FakeAWSCognitoSources) Update(aWSCognitoSource *v1alpha1.AWSCognitoSource) (result *v1alpha1.AWSCognitoSource, err error) {
-	obj, err := c.Fake.
-		Invokes(testing.NewUpdateAction(awscognitosourcesResource, c.ns, aWSCognitoSource), &v1alpha1.AWSCognitoSource{})
-
-	if obj == nil {
-		return nil, err
-	}
-	return obj.(*v1alpha1.AWSCognitoSource), err
-}
-
-// UpdateStatus was generated because the type contains a Status member.
-// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
-func (c *FakeAWSCognitoSources) UpdateStatus(aWSCognitoSource *v1alpha1.AWSCognitoSource) (*v1alpha1.AWSCognitoSource, error) {
-	obj, err := c.Fake.
-		Invokes(testing.NewUpdateSubresourceAction(awscognitosourcesResource, "status", c.ns, aWSCognitoSource), &v1alpha1.AWSCognitoSource{})
-
-	if obj == nil {
-		return nil, err
-	}
-	return obj.(*v1alpha1.AWSCognitoSource), err
-}
-
-// Delete takes name of the aWSCognitoSource and deletes it. Returns an error if one occurs.
-func (c *FakeAWSCognitoSources) Delete(name string, options *v1.DeleteOptions) error {
-	_, err := c.Fake.
-		Invokes(testing.NewDeleteAction(awscognitosourcesResource, c.ns, name), &v1alpha1.AWSCognitoSource{})
-
-	return err
-}
-
-// DeleteCollection deletes a collection of objects.
-func (c *FakeAWSCognitoSources) DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error {
-	action := testing.NewDeleteCollectionAction(awscognitosourcesResource, c.ns, listOptions)
-
-	_, err := c.Fake.Invokes(action, &v1alpha1.AWSCognitoSourceList{})
-	return err
-}
-
-// Patch applies the patch and returns the patched aWSCognitoSource.
-func (c *FakeAWSCognitoSources) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1alpha1.AWSCognitoSource, err error) {
-	obj, err := c.Fake.
-		Invokes(testing.NewPatchSubresourceAction(awscognitosourcesResource, c.ns, name, pt, data, subresources...), &v1alpha1.AWSCognitoSource{})
-
-	if obj == nil {
-		return nil, err
-	}
-	return obj.(*v1alpha1.AWSCognitoSource), err
-}
diff --git a/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/fake/fake_sources_client.go b/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/fake/fake_sources_client.go
index 491f9b8..9d425b8 100644
--- a/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/fake/fake_sources_client.go
+++ b/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/fake/fake_sources_client.go
@@ -32,8 +32,8 @@ func (c *FakeSourcesV1alpha1) AWSCodeCommitSources(namespace string) v1alpha1.AW
 	return &FakeAWSCodeCommitSources{c, namespace}
 }
 
-func (c *FakeSourcesV1alpha1) AWSCognitoSources(namespace string) v1alpha1.AWSCognitoSourceInterface {
-	return &FakeAWSCognitoSources{c, namespace}
+func (c *FakeSourcesV1alpha1) AWSCognitoIdentitySources(namespace string) v1alpha1.AWSCognitoIdentitySourceInterface {
+	return &FakeAWSCognitoIdentitySources{c, namespace}
 }
 
 func (c *FakeSourcesV1alpha1) AWSCognitoUserPoolSources(namespace string) v1alpha1.AWSCognitoUserPoolSourceInterface {
diff --git a/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/generated_expansion.go b/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/generated_expansion.go
index 5a64a79..abdbbed 100644
--- a/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/generated_expansion.go
+++ b/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/generated_expansion.go
@@ -20,7 +20,7 @@ package v1alpha1
 
 type AWSCodeCommitSourceExpansion interface{}
 
-type AWSCognitoSourceExpansion interface{}
+type AWSCognitoIdentitySourceExpansion interface{}
 
 type AWSCognitoUserPoolSourceExpansion interface{}
 
diff --git a/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/sources_client.go b/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/sources_client.go
index b76cfaf..2cdf125 100644
--- a/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/sources_client.go
+++ b/pkg/client/generated/clientset/internalclientset/typed/sources/v1alpha1/sources_client.go
@@ -27,7 +27,7 @@ import (
 type SourcesV1alpha1Interface interface {
 	RESTClient() rest.Interface
 	AWSCodeCommitSourcesGetter
-	AWSCognitoSourcesGetter
+	AWSCognitoIdentitySourcesGetter
 	AWSCognitoUserPoolSourcesGetter
 	AWSDynamoDBSourcesGetter
 	AWSIoTSourcesGetter
@@ -45,8 +45,8 @@ func (c *SourcesV1alpha1Client) AWSCodeCommitSources(namespace string) AWSCodeCo
 	return newAWSCodeCommitSources(c, namespace)
 }
 
-func (c *SourcesV1alpha1Client) AWSCognitoSources(namespace string) AWSCognitoSourceInterface {
-	return newAWSCognitoSources(c, namespace)
+func (c *SourcesV1alpha1Client) AWSCognitoIdentitySources(namespace string) AWSCognitoIdentitySourceInterface {
+	return newAWSCognitoIdentitySources(c, namespace)
 }
 
 func (c *SourcesV1alpha1Client) AWSCognitoUserPoolSources(namespace string) AWSCognitoUserPoolSourceInterface {
diff --git a/pkg/client/generated/informers/externalversions/generic.go b/pkg/client/generated/informers/externalversions/generic.go
index 55bc108..0aac397 100644
--- a/pkg/client/generated/informers/externalversions/generic.go
+++ b/pkg/client/generated/informers/externalversions/generic.go
@@ -55,8 +55,8 @@ func (f *sharedInformerFactory) ForResource(resource schema.GroupVersionResource
 	// Group=sources.triggermesh.io, Version=v1alpha1
 	case v1alpha1.SchemeGroupVersion.WithResource("awscodecommitsources"):
 		return &genericInformer{resource: resource.GroupResource(), informer: f.Sources().V1alpha1().AWSCodeCommitSources().Informer()}, nil
-	case v1alpha1.SchemeGroupVersion.WithResource("awscognitosources"):
-		return &genericInformer{resource: resource.GroupResource(), informer: f.Sources().V1alpha1().AWSCognitoSources().Informer()}, nil
+	case v1alpha1.SchemeGroupVersion.WithResource("awscognitoidentitysources"):
+		return &genericInformer{resource: resource.GroupResource(), informer: f.Sources().V1alpha1().AWSCognitoIdentitySources().Informer()}, nil
 	case v1alpha1.SchemeGroupVersion.WithResource("awscognitouserpoolsources"):
 		return &genericInformer{resource: resource.GroupResource(), informer: f.Sources().V1alpha1().AWSCognitoUserPoolSources().Informer()}, nil
 	case v1alpha1.SchemeGroupVersion.WithResource("awsdynamodbsources"):
diff --git a/pkg/client/generated/informers/externalversions/sources/v1alpha1/awscognitosource.go b/pkg/client/generated/informers/externalversions/sources/v1alpha1/awscognitoidentitysource.go
similarity index 55%
rename from pkg/client/generated/informers/externalversions/sources/v1alpha1/awscognitosource.go
rename to pkg/client/generated/informers/externalversions/sources/v1alpha1/awscognitoidentitysource.go
index 9c7f825..b2668d8 100644
--- a/pkg/client/generated/informers/externalversions/sources/v1alpha1/awscognitosource.go
+++ b/pkg/client/generated/informers/externalversions/sources/v1alpha1/awscognitoidentitysource.go
@@ -31,59 +31,59 @@ import (
 	cache "k8s.io/client-go/tools/cache"
 )
 
-// AWSCognitoSourceInformer provides access to a shared informer and lister for
-// AWSCognitoSources.
-type AWSCognitoSourceInformer interface {
+// AWSCognitoIdentitySourceInformer provides access to a shared informer and lister for
+// AWSCognitoIdentitySources.
+type AWSCognitoIdentitySourceInformer interface {
 	Informer() cache.SharedIndexInformer
-	Lister() v1alpha1.AWSCognitoSourceLister
+	Lister() v1alpha1.AWSCognitoIdentitySourceLister
 }
 
-type aWSCognitoSourceInformer struct {
+type aWSCognitoIdentitySourceInformer struct {
 	factory          internalinterfaces.SharedInformerFactory
 	tweakListOptions internalinterfaces.TweakListOptionsFunc
 	namespace        string
 }
 
-// NewAWSCognitoSourceInformer constructs a new informer for AWSCognitoSource type.
+// NewAWSCognitoIdentitySourceInformer constructs a new informer for AWSCognitoIdentitySource type.
 // Always prefer using an informer factory to get a shared informer instead of getting an independent
 // one. This reduces memory footprint and number of connections to the server.
-func NewAWSCognitoSourceInformer(client internalclientset.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
-	return NewFilteredAWSCognitoSourceInformer(client, namespace, resyncPeriod, indexers, nil)
+func NewAWSCognitoIdentitySourceInformer(client internalclientset.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
+	return NewFilteredAWSCognitoIdentitySourceInformer(client, namespace, resyncPeriod, indexers, nil)
 }
 
-// NewFilteredAWSCognitoSourceInformer constructs a new informer for AWSCognitoSource type.
+// NewFilteredAWSCognitoIdentitySourceInformer constructs a new informer for AWSCognitoIdentitySource type.
 // Always prefer using an informer factory to get a shared informer instead of getting an independent
 // one. This reduces memory footprint and number of connections to the server.
-func NewFilteredAWSCognitoSourceInformer(client internalclientset.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
+func NewFilteredAWSCognitoIdentitySourceInformer(client internalclientset.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
 	return cache.NewSharedIndexInformer(
 		&cache.ListWatch{
 			ListFunc: func(options v1.ListOptions) (runtime.Object, error) {
 				if tweakListOptions != nil {
 					tweakListOptions(&options)
 				}
-				return client.SourcesV1alpha1().AWSCognitoSources(namespace).List(options)
+				return client.SourcesV1alpha1().AWSCognitoIdentitySources(namespace).List(options)
 			},
 			WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
 				if tweakListOptions != nil {
 					tweakListOptions(&options)
 				}
-				return client.SourcesV1alpha1().AWSCognitoSources(namespace).Watch(options)
+				return client.SourcesV1alpha1().AWSCognitoIdentitySources(namespace).Watch(options)
 			},
 		},
-		&sourcesv1alpha1.AWSCognitoSource{},
+		&sourcesv1alpha1.AWSCognitoIdentitySource{},
 		resyncPeriod,
 		indexers,
 	)
 }
 
-func (f *aWSCognitoSourceInformer) defaultInformer(client internalclientset.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
-	return NewFilteredAWSCognitoSourceInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
+func (f *aWSCognitoIdentitySourceInformer) defaultInformer(client internalclientset.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
+	return NewFilteredAWSCognitoIdentitySourceInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
 }
 
-func (f *aWSCognitoSourceInformer) Informer() cache.SharedIndexInformer {
-	return f.factory.InformerFor(&sourcesv1alpha1.AWSCognitoSource{}, f.defaultInformer)
+func (f *aWSCognitoIdentitySourceInformer) Informer() cache.SharedIndexInformer {
+	return f.factory.InformerFor(&sourcesv1alpha1.AWSCognitoIdentitySource{}, f.defaultInformer)
 }
 
-func (f *aWSCognitoSourceInformer) Lister() v1alpha1.AWSCognitoSourceLister {
-	return v1alpha1.NewAWSCognitoSourceLister(f.Informer().GetIndexer())
+func (f *aWSCognitoIdentitySourceInformer) Lister() v1alpha1.AWSCognitoIdentitySourceLister {
+	return v1alpha1.NewAWSCognitoIdentitySourceLister(f.Informer().GetIndexer())
 }
diff --git a/pkg/client/generated/informers/externalversions/sources/v1alpha1/interface.go b/pkg/client/generated/informers/externalversions/sources/v1alpha1/interface.go
index 75d9eb3..c7f23d1 100644
--- a/pkg/client/generated/informers/externalversions/sources/v1alpha1/interface.go
+++ b/pkg/client/generated/informers/externalversions/sources/v1alpha1/interface.go
@@ -26,8 +26,8 @@ import (
 type Interface interface {
 	// AWSCodeCommitSources returns a AWSCodeCommitSourceInformer.
 	AWSCodeCommitSources() AWSCodeCommitSourceInformer
-	// AWSCognitoSources returns a AWSCognitoSourceInformer.
-	AWSCognitoSources() AWSCognitoSourceInformer
+	// AWSCognitoIdentitySources returns a AWSCognitoIdentitySourceInformer.
+	AWSCognitoIdentitySources() AWSCognitoIdentitySourceInformer
 	// AWSCognitoUserPoolSources returns a AWSCognitoUserPoolSourceInformer.
 	AWSCognitoUserPoolSources() AWSCognitoUserPoolSourceInformer
 	// AWSDynamoDBSources returns a AWSDynamoDBSourceInformer.
@@ -58,9 +58,9 @@ func (v *version) AWSCodeCommitSources() AWSCodeCommitSourceInformer {
 	return &aWSCodeCommitSourceInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
 }
 
-// AWSCognitoSources returns a AWSCognitoSourceInformer.
-func (v *version) AWSCognitoSources() AWSCognitoSourceInformer {
-	return &aWSCognitoSourceInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
+// AWSCognitoIdentitySources returns a AWSCognitoIdentitySourceInformer.
+func (v *version) AWSCognitoIdentitySources() AWSCognitoIdentitySourceInformer {
+	return &aWSCognitoIdentitySourceInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
 }
 
 // AWSCognitoUserPoolSources returns a AWSCognitoUserPoolSourceInformer.
diff --git a/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitosource/awscognitosource.go b/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitoidentitysource/awscognitoidentitysource.go
similarity index 85%
rename from pkg/client/generated/injection/informers/sources/v1alpha1/awscognitosource/awscognitosource.go
rename to pkg/client/generated/injection/informers/sources/v1alpha1/awscognitoidentitysource/awscognitoidentitysource.go
index 517a9a5..c150bb0 100644
--- a/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitosource/awscognitosource.go
+++ b/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitoidentitysource/awscognitoidentitysource.go
@@ -16,7 +16,7 @@ limitations under the License.
 
 // Code generated by injection-gen. DO NOT EDIT.
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	context "context"
@@ -37,16 +37,16 @@ type Key struct{}
 
 func withInformer(ctx context.Context) (context.Context, controller.Informer) {
 	f := factory.Get(ctx)
-	inf := f.Sources().V1alpha1().AWSCognitoSources()
+	inf := f.Sources().V1alpha1().AWSCognitoIdentitySources()
 	return context.WithValue(ctx, Key{}, inf), inf.Informer()
 }
 
 // Get extracts the typed informer from the context.
-func Get(ctx context.Context) v1alpha1.AWSCognitoSourceInformer {
+func Get(ctx context.Context) v1alpha1.AWSCognitoIdentitySourceInformer {
 	untyped := ctx.Value(Key{})
 	if untyped == nil {
 		logging.FromContext(ctx).Panic(
-			"Unable to fetch github.com/triggermesh/aws-event-sources/pkg/client/generated/informers/externalversions/sources/v1alpha1.AWSCognitoSourceInformer from context.")
+			"Unable to fetch github.com/triggermesh/aws-event-sources/pkg/client/generated/informers/externalversions/sources/v1alpha1.AWSCognitoIdentitySourceInformer from context.")
 	}
-	return untyped.(v1alpha1.AWSCognitoSourceInformer)
+	return untyped.(v1alpha1.AWSCognitoIdentitySourceInformer)
 }
diff --git a/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitosource/fake/fake.go b/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitoidentitysource/fake/fake.go
similarity index 75%
rename from pkg/client/generated/injection/informers/sources/v1alpha1/awscognitosource/fake/fake.go
rename to pkg/client/generated/injection/informers/sources/v1alpha1/awscognitoidentitysource/fake/fake.go
index 4d139bc..46a9ca6 100644
--- a/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitosource/fake/fake.go
+++ b/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitoidentitysource/fake/fake.go
@@ -22,12 +22,12 @@ import (
 	context "context"
 
 	fake "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/informers/factory/fake"
-	awscognitosource "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitosource"
+	awscognitoidentitysource "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitoidentitysource"
 	controller "knative.dev/pkg/controller"
 	injection "knative.dev/pkg/injection"
 )
 
-var Get = awscognitosource.Get
+var Get = awscognitoidentitysource.Get
 
 func init() {
 	injection.Fake.RegisterInformer(withInformer)
@@ -35,6 +35,6 @@ func init() {
 
 func withInformer(ctx context.Context) (context.Context, controller.Informer) {
 	f := fake.Get(ctx)
-	inf := f.Sources().V1alpha1().AWSCognitoSources()
-	return context.WithValue(ctx, awscognitosource.Key{}, inf), inf.Informer()
+	inf := f.Sources().V1alpha1().AWSCognitoIdentitySources()
+	return context.WithValue(ctx, awscognitoidentitysource.Key{}, inf), inf.Informer()
 }
diff --git a/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource/controller.go b/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource/controller.go
similarity index 88%
rename from pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource/controller.go
rename to pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource/controller.go
index 8c6686b..7710a07 100644
--- a/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource/controller.go
+++ b/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource/controller.go
@@ -16,7 +16,7 @@ limitations under the License.
 
 // Code generated by injection-gen. DO NOT EDIT.
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	context "context"
@@ -26,7 +26,7 @@ import (
 
 	internalclientsetscheme "github.com/triggermesh/aws-event-sources/pkg/client/generated/clientset/internalclientset/scheme"
 	client "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/client"
-	awscognitosource "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitosource"
+	awscognitoidentitysource "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitoidentitysource"
 	corev1 "k8s.io/api/core/v1"
 	watch "k8s.io/apimachinery/pkg/watch"
 	scheme "k8s.io/client-go/kubernetes/scheme"
@@ -38,8 +38,8 @@ import (
 )
 
 const (
-	defaultControllerAgentName = "awscognitosource-controller"
-	defaultFinalizerName       = "awscognitosources.sources.triggermesh.io"
+	defaultControllerAgentName = "awscognitoidentitysource-controller"
+	defaultFinalizerName       = "awscognitoidentitysources.sources.triggermesh.io"
 )
 
 // NewImpl returns a controller.Impl that handles queuing and feeding work from
@@ -54,11 +54,11 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		logger.Fatalf("up to one options function is supported, found %d", len(optionsFns))
 	}
 
-	awscognitosourceInformer := awscognitosource.Get(ctx)
+	awscognitoidentitysourceInformer := awscognitoidentitysource.Get(ctx)
 
 	rec := &reconcilerImpl{
 		Client:        client.Get(ctx),
-		Lister:        awscognitosourceInformer.Lister(),
+		Lister:        awscognitoidentitysourceInformer.Lister(),
 		reconciler:    r,
 		finalizerName: defaultFinalizerName,
 	}
diff --git a/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource/reconciler.go b/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource/reconciler.go
similarity index 88%
rename from pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource/reconciler.go
rename to pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource/reconciler.go
index ff55b82..a657e0e 100644
--- a/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource/reconciler.go
+++ b/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource/reconciler.go
@@ -16,7 +16,7 @@ limitations under the License.
 
 // Code generated by injection-gen. DO NOT EDIT.
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	context "context"
@@ -41,35 +41,35 @@ import (
 )
 
 // Interface defines the strongly typed interfaces to be implemented by a
-// controller reconciling v1alpha1.AWSCognitoSource.
+// controller reconciling v1alpha1.AWSCognitoIdentitySource.
 type Interface interface {
-	// ReconcileKind implements custom logic to reconcile v1alpha1.AWSCognitoSource. Any changes
+	// ReconcileKind implements custom logic to reconcile v1alpha1.AWSCognitoIdentitySource. Any changes
 	// to the objects .Status or .Finalizers will be propagated to the stored
 	// object. It is recommended that implementors do not call any update calls
 	// for the Kind inside of ReconcileKind, it is the responsibility of the calling
 	// controller to propagate those properties. The resource passed to ReconcileKind
 	// will always have an empty deletion timestamp.
-	ReconcileKind(ctx context.Context, o *v1alpha1.AWSCognitoSource) reconciler.Event
+	ReconcileKind(ctx context.Context, o *v1alpha1.AWSCognitoIdentitySource) reconciler.Event
 }
 
 // Finalizer defines the strongly typed interfaces to be implemented by a
-// controller finalizing v1alpha1.AWSCognitoSource.
+// controller finalizing v1alpha1.AWSCognitoIdentitySource.
 type Finalizer interface {
-	// FinalizeKind implements custom logic to finalize v1alpha1.AWSCognitoSource. Any changes
+	// FinalizeKind implements custom logic to finalize v1alpha1.AWSCognitoIdentitySource. Any changes
 	// to the objects .Status or .Finalizers will be ignored. Returning a nil or
 	// Normal type reconciler.Event will allow the finalizer to be deleted on
 	// the resource. The resource passed to FinalizeKind will always have a set
 	// deletion timestamp.
-	FinalizeKind(ctx context.Context, o *v1alpha1.AWSCognitoSource) reconciler.Event
+	FinalizeKind(ctx context.Context, o *v1alpha1.AWSCognitoIdentitySource) reconciler.Event
 }
 
-// reconcilerImpl implements controller.Reconciler for v1alpha1.AWSCognitoSource resources.
+// reconcilerImpl implements controller.Reconciler for v1alpha1.AWSCognitoIdentitySource resources.
 type reconcilerImpl struct {
 	// Client is used to write back status updates.
 	Client internalclientset.Interface
 
 	// Listers index properties about resources
-	Lister sourcesv1alpha1.AWSCognitoSourceLister
+	Lister sourcesv1alpha1.AWSCognitoIdentitySourceLister
 
 	// Recorder is an event recorder for recording Event resources to the
 	// Kubernetes API.
@@ -89,7 +89,7 @@ type reconcilerImpl struct {
 // Check that our Reconciler implements controller.Reconciler
 var _ controller.Reconciler = (*reconcilerImpl)(nil)
 
-func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client internalclientset.Interface, lister sourcesv1alpha1.AWSCognitoSourceLister, recorder record.EventRecorder, r Interface, options ...controller.Options) controller.Reconciler {
+func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client internalclientset.Interface, lister sourcesv1alpha1.AWSCognitoIdentitySourceLister, recorder record.EventRecorder, r Interface, options ...controller.Options) controller.Reconciler {
 	// Check the options function input. It should be 0 or 1.
 	if len(options) > 1 {
 		logger.Fatalf("up to one options struct is supported, found %d", len(options))
@@ -138,7 +138,7 @@ func (r *reconcilerImpl) Reconcile(ctx context.Context, key string) error {
 
 	// Get the resource with this namespace/name.
 
-	getter := r.Lister.AWSCognitoSources(namespace)
+	getter := r.Lister.AWSCognitoIdentitySources(namespace)
 
 	original, err := getter.Get(name)
 
@@ -215,13 +215,13 @@ func (r *reconcilerImpl) Reconcile(ctx context.Context, key string) error {
 	return nil
 }
 
-func (r *reconcilerImpl) updateStatus(existing *v1alpha1.AWSCognitoSource, desired *v1alpha1.AWSCognitoSource) error {
+func (r *reconcilerImpl) updateStatus(existing *v1alpha1.AWSCognitoIdentitySource, desired *v1alpha1.AWSCognitoIdentitySource) error {
 	existing = existing.DeepCopy()
 	return reconciler.RetryUpdateConflicts(func(attempts int) (err error) {
 		// The first iteration tries to use the injectionInformer's state, subsequent attempts fetch the latest state via API.
 		if attempts > 0 {
 
-			getter := r.Client.SourcesV1alpha1().AWSCognitoSources(desired.Namespace)
+			getter := r.Client.SourcesV1alpha1().AWSCognitoIdentitySources(desired.Namespace)
 
 			existing, err = getter.Get(desired.Name, metav1.GetOptions{})
 			if err != nil {
@@ -236,7 +236,7 @@ func (r *reconcilerImpl) updateStatus(existing *v1alpha1.AWSCognitoSource, desir
 
 		existing.Status = desired.Status
 
-		updater := r.Client.SourcesV1alpha1().AWSCognitoSources(existing.Namespace)
+		updater := r.Client.SourcesV1alpha1().AWSCognitoIdentitySources(existing.Namespace)
 
 		_, err = updater.UpdateStatus(existing)
 		return err
@@ -246,9 +246,9 @@ func (r *reconcilerImpl) updateStatus(existing *v1alpha1.AWSCognitoSource, desir
 // updateFinalizersFiltered will update the Finalizers of the resource.
 // TODO: this method could be generic and sync all finalizers. For now it only
 // updates defaultFinalizerName or its override.
-func (r *reconcilerImpl) updateFinalizersFiltered(ctx context.Context, resource *v1alpha1.AWSCognitoSource) (*v1alpha1.AWSCognitoSource, error) {
+func (r *reconcilerImpl) updateFinalizersFiltered(ctx context.Context, resource *v1alpha1.AWSCognitoIdentitySource) (*v1alpha1.AWSCognitoIdentitySource, error) {
 
-	getter := r.Lister.AWSCognitoSources(resource.Namespace)
+	getter := r.Lister.AWSCognitoIdentitySources(resource.Namespace)
 
 	actual, err := getter.Get(resource.Name)
 	if err != nil {
@@ -293,7 +293,7 @@ func (r *reconcilerImpl) updateFinalizersFiltered(ctx context.Context, resource
 		return resource, err
 	}
 
-	patcher := r.Client.SourcesV1alpha1().AWSCognitoSources(resource.Namespace)
+	patcher := r.Client.SourcesV1alpha1().AWSCognitoIdentitySources(resource.Namespace)
 
 	resource, err = patcher.Patch(resource.Name, types.MergePatchType, patch)
 	if err != nil {
@@ -306,7 +306,7 @@ func (r *reconcilerImpl) updateFinalizersFiltered(ctx context.Context, resource
 	return resource, err
 }
 
-func (r *reconcilerImpl) setFinalizerIfFinalizer(ctx context.Context, resource *v1alpha1.AWSCognitoSource) (*v1alpha1.AWSCognitoSource, error) {
+func (r *reconcilerImpl) setFinalizerIfFinalizer(ctx context.Context, resource *v1alpha1.AWSCognitoIdentitySource) (*v1alpha1.AWSCognitoIdentitySource, error) {
 	if _, ok := r.reconciler.(Finalizer); !ok {
 		return resource, nil
 	}
@@ -324,7 +324,7 @@ func (r *reconcilerImpl) setFinalizerIfFinalizer(ctx context.Context, resource *
 	return r.updateFinalizersFiltered(ctx, resource)
 }
 
-func (r *reconcilerImpl) clearFinalizer(ctx context.Context, resource *v1alpha1.AWSCognitoSource, reconcileEvent reconciler.Event) (*v1alpha1.AWSCognitoSource, error) {
+func (r *reconcilerImpl) clearFinalizer(ctx context.Context, resource *v1alpha1.AWSCognitoIdentitySource, reconcileEvent reconciler.Event) (*v1alpha1.AWSCognitoIdentitySource, error) {
 	if _, ok := r.reconciler.(Finalizer); !ok {
 		return resource, nil
 	}
diff --git a/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource/stub/controller.go b/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource/stub/controller.go
similarity index 62%
rename from pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource/stub/controller.go
rename to pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource/stub/controller.go
index 758e494..5617acb 100644
--- a/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource/stub/controller.go
+++ b/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource/stub/controller.go
@@ -16,13 +16,13 @@ limitations under the License.
 
 // Code generated by injection-gen. DO NOT EDIT.
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	context "context"
 
-	awscognitosource "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitosource"
-	v1alpha1awscognitosource "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource"
+	awscognitoidentitysource "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitoidentitysource"
+	v1alpha1awscognitoidentitysource "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource"
 	configmap "knative.dev/pkg/configmap"
 	controller "knative.dev/pkg/controller"
 	logging "knative.dev/pkg/logging"
@@ -30,23 +30,23 @@ import (
 
 // TODO: PLEASE COPY AND MODIFY THIS FILE AS A STARTING POINT
 
-// NewController creates a Reconciler for AWSCognitoSource and returns the result of NewImpl.
+// NewController creates a Reconciler for AWSCognitoIdentitySource and returns the result of NewImpl.
 func NewController(
 	ctx context.Context,
 	cmw configmap.Watcher,
 ) *controller.Impl {
 	logger := logging.FromContext(ctx)
 
-	awscognitosourceInformer := awscognitosource.Get(ctx)
+	awscognitoidentitysourceInformer := awscognitoidentitysource.Get(ctx)
 
 	// TODO: setup additional informers here.
 
 	r := &Reconciler{}
-	impl := v1alpha1awscognitosource.NewImpl(ctx, r)
+	impl := v1alpha1awscognitoidentitysource.NewImpl(ctx, r)
 
 	logger.Info("Setting up event handlers.")
 
-	awscognitosourceInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
+	awscognitoidentitysourceInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 
 	// TODO: add additional informer event handlers here.
 
diff --git a/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource/stub/reconciler.go b/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource/stub/reconciler.go
similarity index 72%
rename from pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource/stub/reconciler.go
rename to pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource/stub/reconciler.go
index 81b013a..5adee6b 100644
--- a/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource/stub/reconciler.go
+++ b/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource/stub/reconciler.go
@@ -16,13 +16,13 @@ limitations under the License.
 
 // Code generated by injection-gen. DO NOT EDIT.
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	context "context"
 
 	v1alpha1 "github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
-	awscognitosource "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource"
+	awscognitoidentitysource "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource"
 	v1 "k8s.io/api/core/v1"
 	reconciler "knative.dev/pkg/reconciler"
 )
@@ -30,24 +30,24 @@ import (
 // TODO: PLEASE COPY AND MODIFY THIS FILE AS A STARTING POINT
 
 // newReconciledNormal makes a new reconciler event with event type Normal, and
-// reason AWSCognitoSourceReconciled.
+// reason AWSCognitoIdentitySourceReconciled.
 func newReconciledNormal(namespace, name string) reconciler.Event {
-	return reconciler.NewEvent(v1.EventTypeNormal, "AWSCognitoSourceReconciled", "AWSCognitoSource reconciled: \"%s/%s\"", namespace, name)
+	return reconciler.NewEvent(v1.EventTypeNormal, "AWSCognitoIdentitySourceReconciled", "AWSCognitoIdentitySource reconciled: \"%s/%s\"", namespace, name)
 }
 
-// Reconciler implements controller.Reconciler for AWSCognitoSource resources.
+// Reconciler implements controller.Reconciler for AWSCognitoIdentitySource resources.
 type Reconciler struct {
 	// TODO: add additional requirements here.
 }
 
 // Check that our Reconciler implements Interface
-var _ awscognitosource.Interface = (*Reconciler)(nil)
+var _ awscognitoidentitysource.Interface = (*Reconciler)(nil)
 
 // Optionally check that our Reconciler implements Finalizer
-//var _ awscognitosource.Finalizer = (*Reconciler)(nil)
+//var _ awscognitoidentitysource.Finalizer = (*Reconciler)(nil)
 
 // ReconcileKind implements Interface.ReconcileKind.
-func (r *Reconciler) ReconcileKind(ctx context.Context, o *v1alpha1.AWSCognitoSource) reconciler.Event {
+func (r *Reconciler) ReconcileKind(ctx context.Context, o *v1alpha1.AWSCognitoIdentitySource) reconciler.Event {
 	// TODO: use this if the resource implements InitializeConditions.
 	// o.Status.InitializeConditions()
 
@@ -60,7 +60,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, o *v1alpha1.AWSCognitoSo
 
 // Optionally, use FinalizeKind to add finalizers. FinalizeKind will be called
 // when the resource is deleted.
-//func (r *Reconciler) FinalizeKind(ctx context.Context, o *v1alpha1.AWSCognitoSource) reconciler.Event {
+//func (r *Reconciler) FinalizeKind(ctx context.Context, o *v1alpha1.AWSCognitoIdentitySource) reconciler.Event {
 //	// TODO: add custom finalization logic here.
 //	return nil
 //}
diff --git a/pkg/client/generated/listers/sources/v1alpha1/awscognitoidentitysource.go b/pkg/client/generated/listers/sources/v1alpha1/awscognitoidentitysource.go
new file mode 100644
index 0000000..2a8bf3c
--- /dev/null
+++ b/pkg/client/generated/listers/sources/v1alpha1/awscognitoidentitysource.go
@@ -0,0 +1,94 @@
+/*
+Copyright (c) 2020 TriggerMesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by lister-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	v1alpha1 "github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+)
+
+// AWSCognitoIdentitySourceLister helps list AWSCognitoIdentitySources.
+type AWSCognitoIdentitySourceLister interface {
+	// List lists all AWSCognitoIdentitySources in the indexer.
+	List(selector labels.Selector) (ret []*v1alpha1.AWSCognitoIdentitySource, err error)
+	// AWSCognitoIdentitySources returns an object that can list and get AWSCognitoIdentitySources.
+	AWSCognitoIdentitySources(namespace string) AWSCognitoIdentitySourceNamespaceLister
+	AWSCognitoIdentitySourceListerExpansion
+}
+
+// aWSCognitoIdentitySourceLister implements the AWSCognitoIdentitySourceLister interface.
+type aWSCognitoIdentitySourceLister struct {
+	indexer cache.Indexer
+}
+
+// NewAWSCognitoIdentitySourceLister returns a new AWSCognitoIdentitySourceLister.
+func NewAWSCognitoIdentitySourceLister(indexer cache.Indexer) AWSCognitoIdentitySourceLister {
+	return &aWSCognitoIdentitySourceLister{indexer: indexer}
+}
+
+// List lists all AWSCognitoIdentitySources in the indexer.
+func (s *aWSCognitoIdentitySourceLister) List(selector labels.Selector) (ret []*v1alpha1.AWSCognitoIdentitySource, err error) {
+	err = cache.ListAll(s.indexer, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1alpha1.AWSCognitoIdentitySource))
+	})
+	return ret, err
+}
+
+// AWSCognitoIdentitySources returns an object that can list and get AWSCognitoIdentitySources.
+func (s *aWSCognitoIdentitySourceLister) AWSCognitoIdentitySources(namespace string) AWSCognitoIdentitySourceNamespaceLister {
+	return aWSCognitoIdentitySourceNamespaceLister{indexer: s.indexer, namespace: namespace}
+}
+
+// AWSCognitoIdentitySourceNamespaceLister helps list and get AWSCognitoIdentitySources.
+type AWSCognitoIdentitySourceNamespaceLister interface {
+	// List lists all AWSCognitoIdentitySources in the indexer for a given namespace.
+	List(selector labels.Selector) (ret []*v1alpha1.AWSCognitoIdentitySource, err error)
+	// Get retrieves the AWSCognitoIdentitySource from the indexer for a given namespace and name.
+	Get(name string) (*v1alpha1.AWSCognitoIdentitySource, error)
+	AWSCognitoIdentitySourceNamespaceListerExpansion
+}
+
+// aWSCognitoIdentitySourceNamespaceLister implements the AWSCognitoIdentitySourceNamespaceLister
+// interface.
+type aWSCognitoIdentitySourceNamespaceLister struct {
+	indexer   cache.Indexer
+	namespace string
+}
+
+// List lists all AWSCognitoIdentitySources in the indexer for a given namespace.
+func (s aWSCognitoIdentitySourceNamespaceLister) List(selector labels.Selector) (ret []*v1alpha1.AWSCognitoIdentitySource, err error) {
+	err = cache.ListAllByNamespace(s.indexer, s.namespace, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1alpha1.AWSCognitoIdentitySource))
+	})
+	return ret, err
+}
+
+// Get retrieves the AWSCognitoIdentitySource from the indexer for a given namespace and name.
+func (s aWSCognitoIdentitySourceNamespaceLister) Get(name string) (*v1alpha1.AWSCognitoIdentitySource, error) {
+	obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, errors.NewNotFound(v1alpha1.Resource("awscognitoidentitysource"), name)
+	}
+	return obj.(*v1alpha1.AWSCognitoIdentitySource), nil
+}
diff --git a/pkg/client/generated/listers/sources/v1alpha1/awscognitosource.go b/pkg/client/generated/listers/sources/v1alpha1/awscognitosource.go
deleted file mode 100644
index d39e6c6..0000000
--- a/pkg/client/generated/listers/sources/v1alpha1/awscognitosource.go
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
-Copyright (c) 2020 TriggerMesh Inc.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-   http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-// Code generated by lister-gen. DO NOT EDIT.
-
-package v1alpha1
-
-import (
-	v1alpha1 "github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
-	"k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/client-go/tools/cache"
-)
-
-// AWSCognitoSourceLister helps list AWSCognitoSources.
-type AWSCognitoSourceLister interface {
-	// List lists all AWSCognitoSources in the indexer.
-	List(selector labels.Selector) (ret []*v1alpha1.AWSCognitoSource, err error)
-	// AWSCognitoSources returns an object that can list and get AWSCognitoSources.
-	AWSCognitoSources(namespace string) AWSCognitoSourceNamespaceLister
-	AWSCognitoSourceListerExpansion
-}
-
-// aWSCognitoSourceLister implements the AWSCognitoSourceLister interface.
-type aWSCognitoSourceLister struct {
-	indexer cache.Indexer
-}
-
-// NewAWSCognitoSourceLister returns a new AWSCognitoSourceLister.
-func NewAWSCognitoSourceLister(indexer cache.Indexer) AWSCognitoSourceLister {
-	return &aWSCognitoSourceLister{indexer: indexer}
-}
-
-// List lists all AWSCognitoSources in the indexer.
-func (s *aWSCognitoSourceLister) List(selector labels.Selector) (ret []*v1alpha1.AWSCognitoSource, err error) {
-	err = cache.ListAll(s.indexer, selector, func(m interface{}) {
-		ret = append(ret, m.(*v1alpha1.AWSCognitoSource))
-	})
-	return ret, err
-}
-
-// AWSCognitoSources returns an object that can list and get AWSCognitoSources.
-func (s *aWSCognitoSourceLister) AWSCognitoSources(namespace string) AWSCognitoSourceNamespaceLister {
-	return aWSCognitoSourceNamespaceLister{indexer: s.indexer, namespace: namespace}
-}
-
-// AWSCognitoSourceNamespaceLister helps list and get AWSCognitoSources.
-type AWSCognitoSourceNamespaceLister interface {
-	// List lists all AWSCognitoSources in the indexer for a given namespace.
-	List(selector labels.Selector) (ret []*v1alpha1.AWSCognitoSource, err error)
-	// Get retrieves the AWSCognitoSource from the indexer for a given namespace and name.
-	Get(name string) (*v1alpha1.AWSCognitoSource, error)
-	AWSCognitoSourceNamespaceListerExpansion
-}
-
-// aWSCognitoSourceNamespaceLister implements the AWSCognitoSourceNamespaceLister
-// interface.
-type aWSCognitoSourceNamespaceLister struct {
-	indexer   cache.Indexer
-	namespace string
-}
-
-// List lists all AWSCognitoSources in the indexer for a given namespace.
-func (s aWSCognitoSourceNamespaceLister) List(selector labels.Selector) (ret []*v1alpha1.AWSCognitoSource, err error) {
-	err = cache.ListAllByNamespace(s.indexer, s.namespace, selector, func(m interface{}) {
-		ret = append(ret, m.(*v1alpha1.AWSCognitoSource))
-	})
-	return ret, err
-}
-
-// Get retrieves the AWSCognitoSource from the indexer for a given namespace and name.
-func (s aWSCognitoSourceNamespaceLister) Get(name string) (*v1alpha1.AWSCognitoSource, error) {
-	obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name)
-	if err != nil {
-		return nil, err
-	}
-	if !exists {
-		return nil, errors.NewNotFound(v1alpha1.Resource("awscognitosource"), name)
-	}
-	return obj.(*v1alpha1.AWSCognitoSource), nil
-}
diff --git a/pkg/client/generated/listers/sources/v1alpha1/expansion_generated.go b/pkg/client/generated/listers/sources/v1alpha1/expansion_generated.go
index e65685d..0e91c42 100644
--- a/pkg/client/generated/listers/sources/v1alpha1/expansion_generated.go
+++ b/pkg/client/generated/listers/sources/v1alpha1/expansion_generated.go
@@ -26,13 +26,13 @@ type AWSCodeCommitSourceListerExpansion interface{}
 // AWSCodeCommitSourceNamespaceLister.
 type AWSCodeCommitSourceNamespaceListerExpansion interface{}
 
-// AWSCognitoSourceListerExpansion allows custom methods to be added to
-// AWSCognitoSourceLister.
-type AWSCognitoSourceListerExpansion interface{}
+// AWSCognitoIdentitySourceListerExpansion allows custom methods to be added to
+// AWSCognitoIdentitySourceLister.
+type AWSCognitoIdentitySourceListerExpansion interface{}
 
-// AWSCognitoSourceNamespaceListerExpansion allows custom methods to be added to
-// AWSCognitoSourceNamespaceLister.
-type AWSCognitoSourceNamespaceListerExpansion interface{}
+// AWSCognitoIdentitySourceNamespaceListerExpansion allows custom methods to be added to
+// AWSCognitoIdentitySourceNamespaceLister.
+type AWSCognitoIdentitySourceNamespaceListerExpansion interface{}
 
 // AWSCognitoUserPoolSourceListerExpansion allows custom methods to be added to
 // AWSCognitoUserPoolSourceLister.
diff --git a/pkg/reconciler/awscognitosource/adapter.go b/pkg/reconciler/awscognitoidentitysource/adapter.go
similarity index 91%
rename from pkg/reconciler/awscognitosource/adapter.go
rename to pkg/reconciler/awscognitoidentitysource/adapter.go
index ae76a7b..0e19387 100644
--- a/pkg/reconciler/awscognitosource/adapter.go
+++ b/pkg/reconciler/awscognitoidentitysource/adapter.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	"fmt"
@@ -32,20 +32,20 @@ import (
 	"github.com/triggermesh/aws-event-sources/pkg/reconciler/common/resource"
 )
 
-const adapterName = "awscognitosource"
+const adapterName = "awscognitoidentitysource"
 
 // 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/awscognitosource"`
+	Image string `default:"gcr.io/triggermesh/awscognitoidentitysource"`
 	// Configuration accessor for logging/metrics/tracing
 	configs source.ConfigAccessor
 }
 
 // adapterDeploymentBuilder returns an AdapterDeploymentBuilderFunc for the
 // given source object and adapter config.
-func adapterDeploymentBuilder(src *v1alpha1.AWSCognitoSource, cfg *adapterConfig) common.AdapterDeploymentBuilderFunc {
+func adapterDeploymentBuilder(src *v1alpha1.AWSCognitoIdentitySource, cfg *adapterConfig) common.AdapterDeploymentBuilderFunc {
 	return func(arn arn.ARN, sinkURI *apis.URL) *appsv1.Deployment {
 		name := kmeta.ChildName(fmt.Sprintf("%s-", adapterName), src.Name)
 
diff --git a/pkg/reconciler/awscognitosource/controller.go b/pkg/reconciler/awscognitoidentitysource/controller.go
similarity index 93%
rename from pkg/reconciler/awscognitosource/controller.go
rename to pkg/reconciler/awscognitoidentitysource/controller.go
index 6f9c5f4..8351ceb 100644
--- a/pkg/reconciler/awscognitosource/controller.go
+++ b/pkg/reconciler/awscognitoidentitysource/controller.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	"context"
@@ -26,8 +26,8 @@ import (
 	"knative.dev/pkg/controller"
 
 	"github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
-	informerv1alpha1 "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitosource"
-	reconcilerv1alpha1 "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource"
+	informerv1alpha1 "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitoidentitysource"
+	reconcilerv1alpha1 "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource"
 	"github.com/triggermesh/aws-event-sources/pkg/reconciler/common"
 )
 
@@ -51,7 +51,7 @@ func NewController(
 
 	r.base = common.NewGenericDeploymentReconciler(
 		ctx,
-		(&v1alpha1.AWSCognitoSource{}).GetGroupVersionKind(),
+		(&v1alpha1.AWSCognitoIdentitySource{}).GetGroupVersionKind(),
 		impl.EnqueueKey,
 		impl.EnqueueControllerOf,
 	)
diff --git a/pkg/reconciler/awscognitosource/reconciler.go b/pkg/reconciler/awscognitoidentitysource/reconciler.go
similarity index 93%
rename from pkg/reconciler/awscognitosource/reconciler.go
rename to pkg/reconciler/awscognitoidentitysource/reconciler.go
index 3fce661..bc4188a 100644
--- a/pkg/reconciler/awscognitosource/reconciler.go
+++ b/pkg/reconciler/awscognitoidentitysource/reconciler.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	"context"
@@ -22,7 +22,7 @@ import (
 	"knative.dev/pkg/reconciler"
 
 	"github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
-	reconcilerv1alpha1 "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource"
+	reconcilerv1alpha1 "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource"
 	"github.com/triggermesh/aws-event-sources/pkg/reconciler/common"
 )
 
@@ -36,7 +36,7 @@ type Reconciler struct {
 var _ reconcilerv1alpha1.Interface = (*Reconciler)(nil)
 
 // ReconcileKind implements Interface.ReconcileKind.
-func (r *Reconciler) ReconcileKind(ctx context.Context, src *v1alpha1.AWSCognitoSource) reconciler.Event {
+func (r *Reconciler) ReconcileKind(ctx context.Context, src *v1alpha1.AWSCognitoIdentitySource) reconciler.Event {
 	// inject source into context for usage in reconciliation logic
 	ctx = v1alpha1.WithSource(ctx, src)
 
diff --git a/pkg/reconciler/awssnssource/adapter.go b/pkg/reconciler/awssnssource/adapter.go
index 1f0481b..0478876 100644
--- a/pkg/reconciler/awssnssource/adapter.go
+++ b/pkg/reconciler/awssnssource/adapter.go
@@ -17,15 +17,15 @@ limitations under the License.
 package awssnssource
 
 import (
+	"encoding/json"
 	"fmt"
 
 	"github.com/aws/aws-sdk-go/aws/arn"
 
-	appsv1 "k8s.io/api/apps/v1"
-
 	"knative.dev/eventing/pkg/reconciler/source"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/kmeta"
+	servingv1 "knative.dev/serving/pkg/apis/serving/v1"
 
 	"github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
 	"github.com/triggermesh/aws-event-sources/pkg/reconciler/common"
@@ -34,19 +34,30 @@ import (
 
 const adapterName = "awssnssource"
 
+const (
+	envSubscriptionAttrs = "SUBSCRIPTION_ATTRIBUTES"
+	envPublicURL         = "PUBLIC_URL"
+)
+
 // 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/awssnssource"`
+
+	// Public domain used to expose Knative Services
+	Domain string `envconfig:"KNATIVE_DOMAIN" default:"example.com"`
+	// URL scheme used to expose public Knative Services
+	Scheme string `envconfig:"KNATIVE_URL_SCHEME" default:"http"`
+
 	// Configuration accessor for logging/metrics/tracing
 	configs source.ConfigAccessor
 }
 
-// adapterDeploymentBuilder returns an AdapterDeploymentBuilderFunc for the
+// adapterServiceBuilder returns an AdapterServiceBuilderFunc for the
 // given source object and adapter config.
-func adapterDeploymentBuilder(src *v1alpha1.AWSSNSSource, cfg *adapterConfig) common.AdapterDeploymentBuilderFunc {
-	return func(arn arn.ARN, sinkURI *apis.URL) *appsv1.Deployment {
+func adapterServiceBuilder(src *v1alpha1.AWSSNSSource, cfg *adapterConfig) common.AdapterServiceBuilderFunc {
+	return func(arn arn.ARN, sinkURI *apis.URL) *servingv1.Service {
 		name := kmeta.ChildName(fmt.Sprintf("%s-", adapterName), src.Name)
 
 		var sinkURIStr string
@@ -54,7 +65,18 @@ func adapterDeploymentBuilder(src *v1alpha1.AWSSNSSource, cfg *adapterConfig) co
 			sinkURIStr = sinkURI.String()
 		}
 
-		return resource.NewDeployment(src.Namespace, name,
+		var subsAttrsStr string
+		if subsAttrsJSON, err := json.Marshal(src.Spec.SubscriptionAttributes); err == nil {
+			subsAttrsStr = string(subsAttrsJSON)
+		}
+
+		/* FIXME(antoineco): we wouldn't need to know the service URL in
+		   advance if we could reconcile SNS subscriptions from the controller.
+		   Ref. https://github.com/triggermesh/aws-event-sources/issues/157
+		*/
+		adapterURL := fmt.Sprintf("%s://%s.%s.%s", cfg.Scheme, name, src.Namespace, cfg.Domain)
+
+		return resource.NewKnService(src.Namespace, name,
 			resource.Controller(src),
 
 			resource.Label(common.AppNameLabel, adapterName),
@@ -63,8 +85,8 @@ func adapterDeploymentBuilder(src *v1alpha1.AWSSNSSource, cfg *adapterConfig) co
 			resource.Label(common.AppPartOfLabel, common.PartOf),
 			resource.Label(common.AppManagedByLabel, common.ManagedBy),
 
-			resource.Selector(common.AppNameLabel, adapterName),
-			resource.Selector(common.AppInstanceLabel, src.Name),
+			resource.PodLabel(common.AppNameLabel, adapterName),
+			resource.PodLabel(common.AppInstanceLabel, src.Name),
 			resource.PodLabel(common.AppComponentLabel, common.AdapterComponent),
 			resource.PodLabel(common.AppPartOfLabel, common.PartOf),
 			resource.PodLabel(common.AppManagedByLabel, common.ManagedBy),
@@ -75,6 +97,8 @@ func adapterDeploymentBuilder(src *v1alpha1.AWSSNSSource, cfg *adapterConfig) co
 			resource.EnvVar(common.EnvNamespace, src.Namespace),
 			resource.EnvVar(common.EnvSink, sinkURIStr),
 			resource.EnvVar(common.EnvARN, arn.String()),
+			resource.EnvVar(envSubscriptionAttrs, subsAttrsStr),
+			resource.EnvVar(envPublicURL, adapterURL),
 			resource.EnvVarFromSecret(common.EnvAccessKeyID,
 				src.Spec.Credentials.AccessKeyID.ValueFromSecret.Name,
 				src.Spec.Credentials.AccessKeyID.ValueFromSecret.Key),
@@ -82,6 +106,9 @@ func adapterDeploymentBuilder(src *v1alpha1.AWSSNSSource, cfg *adapterConfig) co
 				src.Spec.Credentials.SecretAccessKey.ValueFromSecret.Name,
 				src.Spec.Credentials.SecretAccessKey.ValueFromSecret.Key),
 			resource.EnvVars(cfg.configs.ToEnvVars()...),
+			/* FIXME(antoineco): default metrics port 9090 overlaps with queue-proxy
+			 */
+			resource.EnvVar(source.EnvMetricsCfg, ""),
 		)
 	}
 }
diff --git a/pkg/reconciler/awssnssource/controller.go b/pkg/reconciler/awssnssource/controller.go
index ac67803..8b67726 100644
--- a/pkg/reconciler/awssnssource/controller.go
+++ b/pkg/reconciler/awssnssource/controller.go
@@ -49,7 +49,7 @@ func NewController(
 	}
 	impl := reconcilerv1alpha1.NewImpl(ctx, r)
 
-	r.base = common.NewGenericDeploymentReconciler(
+	r.base = common.NewGenericServiceReconciler(
 		ctx,
 		(&v1alpha1.AWSSNSSource{}).GetGroupVersionKind(),
 		impl.EnqueueKey,
diff --git a/pkg/reconciler/awssnssource/reconciler.go b/pkg/reconciler/awssnssource/reconciler.go
index 6f2923d..ec1c353 100644
--- a/pkg/reconciler/awssnssource/reconciler.go
+++ b/pkg/reconciler/awssnssource/reconciler.go
@@ -28,7 +28,7 @@ import (
 
 // Reconciler implements controller.Reconciler for the event source type.
 type Reconciler struct {
-	base       common.GenericDeploymentReconciler
+	base       common.GenericServiceReconciler
 	adapterCfg *adapterConfig
 }
 
@@ -36,9 +36,12 @@ type Reconciler struct {
 var _ reconcilerv1alpha1.Interface = (*Reconciler)(nil)
 
 // ReconcileKind implements Interface.ReconcileKind.
+/* TODO(antoineco): reconcile SNS subscriptions.
+   Ref. https://github.com/triggermesh/aws-event-sources/issues/157
+*/
 func (r *Reconciler) ReconcileKind(ctx context.Context, src *v1alpha1.AWSSNSSource) reconciler.Event {
 	// inject source into context for usage in reconciliation logic
 	ctx = v1alpha1.WithSource(ctx, src)
 
-	return r.base.ReconcileSource(ctx, adapterDeploymentBuilder(src, r.adapterCfg))
+	return r.base.ReconcileSource(ctx, adapterServiceBuilder(src, r.adapterCfg))
 }
diff --git a/pkg/reconciler/common/base.go b/pkg/reconciler/common/base.go
index c6d0537..6cdbb0c 100644
--- a/pkg/reconciler/common/base.go
+++ b/pkg/reconciler/common/base.go
@@ -29,6 +29,10 @@ import (
 	deploymentinformerv1 "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/resolver"
+	servingclientv1 "knative.dev/serving/pkg/client/clientset/versioned/typed/serving/v1"
+	servingclient "knative.dev/serving/pkg/client/injection/client"
+	serviceinformerv1 "knative.dev/serving/pkg/client/injection/informers/serving/v1/service"
+	servinglistersv1 "knative.dev/serving/pkg/client/listers/serving/v1"
 )
 
 // GenericDeploymentReconciler contains interfaces shared across Deployment reconcilers.
@@ -41,6 +45,16 @@ type GenericDeploymentReconciler struct {
 	Lister func(namespace string) appslistersv1.DeploymentNamespaceLister
 }
 
+// GenericServiceReconciler contains interfaces shared across Service reconcilers.
+type GenericServiceReconciler struct {
+	// URI resolver for sinks
+	SinkResolver *resolver.URIResolver
+	// API clients
+	Client func(namespace string) servingclientv1.ServiceInterface
+	// objects listers
+	Lister func(namespace string) servinglistersv1.ServiceNamespaceLister
+}
+
 // NewGenericDeploymentReconciler creates a new GenericDeploymentReconciler and
 // attaches a default event handler to its Deployment informer.
 func NewGenericDeploymentReconciler(ctx context.Context, gvk schema.GroupVersionKind,
@@ -63,3 +77,26 @@ func NewGenericDeploymentReconciler(ctx context.Context, gvk schema.GroupVersion
 
 	return r
 }
+
+// NewGenericServiceReconciler creates a new GenericServiceReconciler and
+// attaches a default event handler to its Service informer.
+func NewGenericServiceReconciler(ctx context.Context, gvk schema.GroupVersionKind,
+	resolverCallback func(types.NamespacedName),
+	adapterHandlerFn func(obj interface{}),
+) GenericServiceReconciler {
+
+	informer := serviceinformerv1.Get(ctx)
+
+	r := GenericServiceReconciler{
+		SinkResolver: resolver.NewURIResolver(ctx, resolverCallback),
+		Client:       servingclient.Get(ctx).ServingV1().Services,
+		Lister:       informer.Lister().Services,
+	}
+
+	informer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: controller.FilterControllerGVK(gvk),
+		Handler:    controller.HandleAll(adapterHandlerFn),
+	})
+
+	return r
+}
diff --git a/pkg/reconciler/common/reconcile.go b/pkg/reconciler/common/reconcile.go
index fadfb1b..5c1f21c 100644
--- a/pkg/reconciler/common/reconcile.go
+++ b/pkg/reconciler/common/reconcile.go
@@ -23,10 +23,13 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/reconciler"
+	"knative.dev/serving/pkg/apis/serving"
+	servingv1 "knative.dev/serving/pkg/apis/serving/v1"
 
 	"github.com/aws/aws-sdk-go/aws/arn"
 
@@ -35,6 +38,13 @@ import (
 	"github.com/triggermesh/aws-event-sources/pkg/reconciler/common/semantic"
 )
 
+// List of annotations set on Knative Serving objects by the Knative Serving
+// admission webhook.
+var knativeServingAnnotations = []string{
+	serving.CreatorAnnotation,
+	serving.UpdaterAnnotation,
+}
+
 // AdapterDeploymentBuilderFunc builds a Deployment object for a source's adapter.
 type AdapterDeploymentBuilderFunc func(arn arn.ARN, sinkURI *apis.URL) *appsv1.Deployment
 
@@ -144,3 +154,120 @@ func (r *GenericDeploymentReconciler) syncAdapterDeployment(ctx context.Context,
 
 	return adapter, nil
 }
+
+// AdapterServiceBuilderFunc builds a Service object for a source's adapter.
+type AdapterServiceBuilderFunc func(arn arn.ARN, sinkURI *apis.URL) *servingv1.Service
+
+// ReconcileSource reconciles an event source type.
+func (r *GenericServiceReconciler) ReconcileSource(ctx context.Context, adb AdapterServiceBuilderFunc) reconciler.Event {
+	src := v1alpha1.SourceFromContext(ctx)
+
+	src.GetStatus().InitializeConditions()
+	src.GetStatus().ObservedGeneration = src.GetGeneration()
+
+	arn, err := arn.Parse(src.GetARN())
+	if err != nil {
+		return controller.NewPermanentError(reconciler.NewEvent(corev1.EventTypeWarning,
+			ReasonInvalidSpec, "Failed to parse ARN: %s", err))
+	}
+	src.GetStatus().CloudEventAttributes = CreateCloudEventAttributes(arn, src.GetEventTypes())
+
+	sinkURI, err := r.resolveSinkURL(ctx)
+	if err != nil {
+		src.GetStatus().MarkNoSink()
+		return controller.NewPermanentError(reconciler.NewEvent(corev1.EventTypeWarning,
+			ReasonBadSinkURI, "Could not resolve sink URI: %s", err))
+	}
+	src.GetStatus().MarkSink(sinkURI)
+
+	if err := r.reconcileAdapter(ctx, adb(arn, sinkURI)); err != nil {
+		return fmt.Errorf("failed to reconcile adapter: %w", err)
+	}
+	return nil
+}
+
+// resolveSinkURL resolves the URL of a sink reference.
+func (r *GenericServiceReconciler) resolveSinkURL(ctx context.Context) (*apis.URL, error) {
+	src := v1alpha1.SourceFromContext(ctx)
+	sink := *src.GetSink()
+
+	if sinkRef := &src.GetSink().Ref; *sinkRef != nil && (*sinkRef).Namespace == "" {
+		(*sinkRef).Namespace = src.GetNamespace()
+	}
+
+	return r.SinkResolver.URIFromDestinationV1(sink, src)
+}
+
+// reconcileAdapter reconciles the state of the source's adapter.
+func (r *GenericServiceReconciler) reconcileAdapter(ctx context.Context, desiredAdapter *servingv1.Service) error {
+	src := v1alpha1.SourceFromContext(ctx)
+
+	currentAdapter, err := r.getOrCreateAdapter(ctx, desiredAdapter)
+	if err != nil {
+		src.GetStatus().PropagateAvailability(currentAdapter)
+		return err
+	}
+
+	currentAdapter, err = r.syncAdapterService(ctx, currentAdapter, desiredAdapter)
+	if err != nil {
+		return fmt.Errorf("failed to synchronize adapter Service: %w", err)
+	}
+	src.GetStatus().PropagateAvailability(currentAdapter)
+
+	return nil
+}
+
+// getOrCreateAdapter returns the existing adapter Service for a given
+// source, or creates it if it is missing.
+func (r *GenericServiceReconciler) getOrCreateAdapter(ctx context.Context, desiredAdapter *servingv1.Service) (*servingv1.Service, error) {
+	src := v1alpha1.SourceFromContext(ctx)
+
+	adapter, err := r.Lister(src.GetNamespace()).Get(desiredAdapter.Name)
+	switch {
+	case apierrors.IsNotFound(err):
+		adapter, err = r.Client(src.GetNamespace()).Create(desiredAdapter)
+		if err != nil {
+			return nil, reconciler.NewEvent(corev1.EventTypeWarning, ReasonFailedAdapterCreate,
+				"Failed to create adapter Service %q: %s", desiredAdapter.Name, err)
+		}
+		event.Normal(ctx, ReasonAdapterCreate, "Created adapter Service %q", adapter.Name)
+
+	case err != nil:
+		return nil, fmt.Errorf("failed to get adapter Service from cache: %w", err)
+	}
+
+	return adapter, nil
+}
+
+// syncAdapterService synchronizes the desired state of an adapter Service
+// against its current state in the running cluster.
+func (r *GenericServiceReconciler) syncAdapterService(ctx context.Context,
+	currentAdapter, desiredAdapter *servingv1.Service) (*servingv1.Service, error) {
+
+	if semantic.Semantic.DeepEqual(desiredAdapter, currentAdapter) {
+		return currentAdapter, nil
+	}
+
+	// resourceVersion must be returned to the API server unmodified for
+	// optimistic concurrency, as per Kubernetes API conventions
+	desiredAdapter.ResourceVersion = currentAdapter.ResourceVersion
+
+	// immutable Knative annotations must be preserved
+	for _, ann := range knativeServingAnnotations {
+		if val, ok := currentAdapter.Annotations[ann]; ok {
+			metav1.SetMetaDataAnnotation(&desiredAdapter.ObjectMeta, ann, val)
+		}
+	}
+
+	// (fake Clientset) preserve status to avoid resetting conditions
+	desiredAdapter.Status = currentAdapter.Status
+
+	adapter, err := r.Client(currentAdapter.Namespace).Update(desiredAdapter)
+	if err != nil {
+		return nil, reconciler.NewEvent(corev1.EventTypeWarning, ReasonFailedAdapterUpdate,
+			"Failed to update adapter Service %q: %s", desiredAdapter.Name, err)
+	}
+	event.Normal(ctx, ReasonAdapterUpdate, "Updated adapter Service %q", adapter.Name)
+
+	return adapter, nil
+}
diff --git a/pkg/reconciler/common/resource/container.go b/pkg/reconciler/common/resource/container.go
index 3cb13b6..0d7cbae 100644
--- a/pkg/reconciler/common/resource/container.go
+++ b/pkg/reconciler/common/resource/container.go
@@ -20,8 +20,12 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
+
+	servingv1 "knative.dev/serving/pkg/apis/serving/v1"
 )
 
+const defaultContainerName = "adapter"
+
 // NewContainer creates a Container object.
 func NewContainer(name string, opts ...ObjectOption) *corev1.Container {
 	c := &corev1.Container{
@@ -41,10 +45,10 @@ func Image(img string) ObjectOption {
 		var image *string
 
 		switch o := object.(type) {
-		case *appsv1.Deployment:
-			image = &firstContainer(o).Image
 		case *corev1.Container:
 			image = &o.Image
+		case *appsv1.Deployment, *servingv1.Service:
+			image = &firstContainer(o).Image
 		}
 
 		*image = img
@@ -59,14 +63,24 @@ func Port(name string, port int32) ObjectOption {
 		switch o := object.(type) {
 		case *corev1.Container:
 			ports = &o.Ports
-		case *appsv1.Deployment:
+		case *appsv1.Deployment, *servingv1.Service:
 			ports = &firstContainer(o).Ports
 		}
 
-		*ports = append(*ports, corev1.ContainerPort{
-			Name:          name,
-			ContainerPort: port,
-		})
+		switch object.(type) {
+		case *corev1.Container, *appsv1.Deployment:
+			*ports = append(*ports, corev1.ContainerPort{
+				Name:          name,
+				ContainerPort: port,
+			})
+
+		case *servingv1.Service:
+			// Knative Services can only define 1 port
+			*ports = []corev1.ContainerPort{{
+				Name:          name,
+				ContainerPort: port,
+			}}
+		}
 	}
 }
 
@@ -106,7 +120,7 @@ func envVarsFrom(object interface{}) (envVars *[]corev1.EnvVar) {
 	switch o := object.(type) {
 	case *corev1.Container:
 		envVars = &o.Env
-	case *appsv1.Deployment:
+	case *appsv1.Deployment, *servingv1.Service:
 		envVars = &firstContainer(o).Env
 	}
 
@@ -125,29 +139,45 @@ func setEnvVar(envVars *[]corev1.EnvVar, name, value string, valueFrom *corev1.E
 func Probe(path, port string) ObjectOption {
 	return func(object interface{}) {
 		var rp **corev1.Probe
+		var intstrPort intstr.IntOrString
 
 		switch o := object.(type) {
 		case *corev1.Container:
 			rp = &o.ReadinessProbe
-		case *appsv1.Deployment:
+		case *appsv1.Deployment, *servingv1.Service:
 			rp = &firstContainer(o).ReadinessProbe
 		}
 
+		switch object.(type) {
+		case *corev1.Container, *appsv1.Deployment:
+			intstrPort = intstr.FromString(port)
+		case *servingv1.Service:
+			// setting port explicitly is invalid in a Knative Service
+		}
+
 		*rp = &corev1.Probe{
 			Handler: corev1.Handler{
 				HTTPGet: &corev1.HTTPGetAction{
 					Path: path,
-					Port: intstr.FromString(port),
+					Port: intstrPort,
 				},
 			},
 		}
 	}
 }
 
-// firstContainer returns a Deployment's first Container definition.
-// A new empty Container is injected if the Deployment does not contain any.
-func firstContainer(d *appsv1.Deployment) *corev1.Container {
-	containers := &d.Spec.Template.Spec.Containers
+// firstContainer returns a PodSpecable's first Container definition.
+// A new empty Container is injected if the PodSpecable does not contain any.
+func firstContainer(object interface{}) *corev1.Container {
+	var containers *[]corev1.Container
+
+	switch o := object.(type) {
+	case *appsv1.Deployment:
+		containers = &o.Spec.Template.Spec.Containers
+	case *servingv1.Service:
+		containers = &o.Spec.Template.Spec.Containers
+	}
+
 	if len(*containers) == 0 {
 		*containers = make([]corev1.Container, 1)
 	}
diff --git a/pkg/reconciler/common/resource/deployment.go b/pkg/reconciler/common/resource/deployment.go
index eb38cdd..e7f48a1 100644
--- a/pkg/reconciler/common/resource/deployment.go
+++ b/pkg/reconciler/common/resource/deployment.go
@@ -18,12 +18,9 @@ package resource
 
 import (
 	appsv1 "k8s.io/api/apps/v1"
-	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-const defaultContainerName = "adapter"
-
 // NewDeployment creates a Deployment object.
 func NewDeployment(ns, name string, opts ...ObjectOption) *appsv1.Deployment {
 	d := &appsv1.Deployment{
@@ -63,22 +60,3 @@ func Selector(key, val string) ObjectOption {
 		PodLabel(key, val)(d)
 	}
 }
-
-// PodLabel sets the value of a label of a Deployment's Pod template.
-func PodLabel(key, val string) ObjectOption {
-	return func(object interface{}) {
-		d := object.(*appsv1.Deployment)
-
-		Label(key, val)(&d.Spec.Template)
-	}
-}
-
-// Container adds a container to a Deployment's Pod template.
-func Container(c *corev1.Container) ObjectOption {
-	return func(object interface{}) {
-		d := object.(*appsv1.Deployment)
-
-		containers := &d.Spec.Template.Spec.Containers
-		*containers = append(*containers, *c)
-	}
-}
diff --git a/pkg/reconciler/common/resource/knservice.go b/pkg/reconciler/common/resource/knservice.go
new file mode 100644
index 0000000..221d146
--- /dev/null
+++ b/pkg/reconciler/common/resource/knservice.go
@@ -0,0 +1,44 @@
+/*
+Copyright (c) 2020 TriggerMesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package resource
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	servingv1 "knative.dev/serving/pkg/apis/serving/v1"
+)
+
+// NewKnService creates a Knative Service object.
+func NewKnService(ns, name string, opts ...ObjectOption) *servingv1.Service {
+	d := &servingv1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: ns,
+			Name:      name,
+		},
+	}
+
+	for _, opt := range opts {
+		opt(d)
+	}
+
+	// ensure the container name is not empty
+	containers := d.Spec.Template.Spec.Containers
+	if len(containers) == 1 && containers[0].Name == "" {
+		containers[0].Name = defaultContainerName
+	}
+
+	return d
+}
diff --git a/pkg/reconciler/common/resource/podspecable.go b/pkg/reconciler/common/resource/podspecable.go
new file mode 100644
index 0000000..0ec1bd2
--- /dev/null
+++ b/pkg/reconciler/common/resource/podspecable.go
@@ -0,0 +1,55 @@
+/*
+Copyright (c) 2020 TriggerMesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package resource
+
+import (
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	servingv1 "knative.dev/serving/pkg/apis/serving/v1"
+)
+
+// PodLabel sets the value of a label of a PodSpecable's Pod template.
+func PodLabel(key, val string) ObjectOption {
+	return func(object interface{}) {
+		var metaObj metav1.Object
+
+		switch o := object.(type) {
+		case *appsv1.Deployment:
+			metaObj = &o.Spec.Template
+		case *servingv1.Service:
+			metaObj = &o.Spec.Template
+		}
+
+		Label(key, val)(metaObj)
+	}
+}
+
+// Container adds a container to a PodSpecable's Pod template.
+func Container(c *corev1.Container) ObjectOption {
+	return func(object interface{}) {
+		switch o := object.(type) {
+		case *appsv1.Deployment:
+			containers := &o.Spec.Template.Spec.Containers
+			*containers = append(*containers, *c)
+		case *servingv1.Service:
+			containers := &o.Spec.Template.Spec.Containers
+			*containers = []corev1.Container{*c}
+		}
+	}
+}
diff --git a/pkg/reconciler/common/semantic/semantic.go b/pkg/reconciler/common/semantic/semantic.go
index 85a8580..c207d91 100644
--- a/pkg/reconciler/common/semantic/semantic.go
+++ b/pkg/reconciler/common/semantic/semantic.go
@@ -21,6 +21,8 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/conversion"
+
+	servingv1 "knative.dev/serving/pkg/apis/serving/v1"
 )
 
 // Semantic can do semantic deep equality checks for Kubernetes API objects.
@@ -33,6 +35,7 @@ import (
 // DeepDerivative comparisons to work as expected.
 var Semantic = conversion.EqualitiesOrDie(
 	deploymentEqual,
+	knServiceEqual,
 )
 
 // eq is an instance of Equalities for internal deep derivative comparisons
@@ -77,3 +80,23 @@ func deploymentEqual(a, b *appsv1.Deployment) bool {
 
 	return true
 }
+
+// knServiceEqual returns whether two Knative Services are semantically equivalent.
+func knServiceEqual(a, b *servingv1.Service) bool {
+	if a == b {
+		return true
+	}
+	if a == nil || b == nil {
+		return false
+	}
+
+	if !eq.DeepDerivative(&a.ObjectMeta, &b.ObjectMeta) {
+		return false
+	}
+
+	if !eq.DeepDerivative(&a.Spec.Template, &b.Spec.Template) {
+		return false
+	}
+
+	return true
+}


Test patch:
diff --git a/pkg/adapter/awscognitosource/adapter_test.go b/pkg/adapter/awscognitoidentitysource/adapter_test.go
similarity index 99%
rename from pkg/adapter/awscognitosource/adapter_test.go
rename to pkg/adapter/awscognitoidentitysource/adapter_test.go
index 7f895a5..88e0ce0 100644
--- a/pkg/adapter/awscognitosource/adapter_test.go
+++ b/pkg/adapter/awscognitoidentitysource/adapter_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	"errors"
diff --git a/pkg/adapter/awssnssource/adapter_test.go b/pkg/adapter/awssnssource/adapter_test.go
index 91eae25..b951986 100644
--- a/pkg/adapter/awssnssource/adapter_test.go
+++ b/pkg/adapter/awssnssource/adapter_test.go
@@ -18,11 +18,14 @@ package awssnssource
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -37,18 +40,11 @@ import (
 type mockedSNSClient struct {
 	snsiface.SNSAPI
 
-	confirmSubsOutput *sns.ConfirmSubscriptionOutput
-	confirmSubsError  error
-
-	createTopicOutput      *sns.CreateTopicOutput
-	createTopicOutputError error
-
 	subscribeOutput      *sns.SubscribeOutput
 	subscribeOutputError error
-}
 
-func (m mockedSNSClient) CreateTopic(_ *sns.CreateTopicInput) (*sns.CreateTopicOutput, error) {
-	return m.createTopicOutput, m.createTopicOutputError
+	confirmSubsOutput *sns.ConfirmSubscriptionOutput
+	confirmSubsError  error
 }
 
 func (m mockedSNSClient) Subscribe(_ *sns.SubscribeInput) (*sns.SubscribeOutput, error) {
@@ -59,7 +55,39 @@ func (m mockedSNSClient) ConfirmSubscription(_ *sns.ConfirmSubscriptionInput) (*
 	return m.confirmSubsOutput, m.confirmSubsError
 }
 
-func TestHandleNotification(t *testing.T) {
+// TestStart verifies that a started adapter responds to cancelation.
+func TestStart(t *testing.T) {
+	const testTimeout = time.Second * 2
+	testCtx, testCancel := context.WithTimeout(context.Background(), testTimeout)
+	defer testCancel()
+
+	a := &adapter{
+		logger: loggingtesting.TestLogger(t),
+	}
+
+	// errCh receives the error value returned by the receiver after
+	// termination. We leave it open to avoid panicking in case the
+	// receiver returns after the timeout.
+	errCh := make(chan error)
+
+	// ctx gets canceled to cause a voluntary interruption of the receiver
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		errCh <- a.Start(ctx.Done())
+	}()
+	cancel()
+
+	select {
+	case <-testCtx.Done():
+		t.Errorf("Test timed out after %v", testTimeout)
+	case err := <-errCh:
+		if err != nil {
+			t.Errorf("Receiver returned an error: %s", err)
+		}
+	}
+}
+
+func TestHandler(t *testing.T) {
 	ceClient := adaptertest.NewTestClient()
 
 	a := &adapter{
@@ -71,6 +99,8 @@ func TestHandleNotification(t *testing.T) {
 		confirmSubsOutput: &sns.ConfirmSubscriptionOutput{SubscriptionArn: aws.String("fooArn")},
 	}
 
+	// handle subscribe
+
 	data, err := ioutil.ReadFile("testSNSConfirmSubscriptionEvent.json")
 	if err != nil {
 		t.Fatalf("Failed to read test file: %v", err)
@@ -87,6 +117,11 @@ func TestHandleNotification(t *testing.T) {
 	handler.ServeHTTP(rr, req)
 	assert.Equal(t, http.StatusOK, rr.Code)
 
+	gotEvents := ceClient.Sent()
+	assert.Len(t, gotEvents, 0, "Expect no event")
+
+	// handle notification
+
 	data, err = ioutil.ReadFile("testSNSNotificationEvent.json")
 	if err != nil {
 		t.Fatalf("Failed to open test file: %v", err)
@@ -102,37 +137,34 @@ func TestHandleNotification(t *testing.T) {
 
 	handler.ServeHTTP(rr, req)
 	assert.Equal(t, http.StatusOK, rr.Code)
+
+	gotEvents = ceClient.Sent()
+	assert.Len(t, gotEvents, 1, "Expect a single event")
+
+	gotData := gotEvents[0].Data()
+	assert.EqualValues(t, data, gotData, "Received event data should equal sent payload")
 }
 
-func TestAttempSubscription(t *testing.T) {
+func TestReconcileSubscription(t *testing.T) {
 	a := &adapter{
-		logger: loggingtesting.TestLogger(t),
-	}
-
-	a.snsClient = mockedSNSClient{
-		createTopicOutput:      &sns.CreateTopicOutput{},
-		createTopicOutputError: errors.New("fake error"),
+		logger:    loggingtesting.TestLogger(t),
+		publicURL: url.URL{Scheme: "http", Host: "example.com"},
 	}
 
-	err := a.attempSubscription(0)
-	assert.Error(t, err)
-
 	a.snsClient = mockedSNSClient{
-		createTopicOutput:    &sns.CreateTopicOutput{TopicArn: aws.String("fooArn")},
 		subscribeOutput:      &sns.SubscribeOutput{},
 		subscribeOutputError: errors.New("fake error"),
 	}
 
-	err = a.attempSubscription(0)
+	err := a.reconcileSubscription()
 	assert.Error(t, err)
 
 	a.snsClient = mockedSNSClient{
-		createTopicOutput:    &sns.CreateTopicOutput{TopicArn: aws.String("fooArn")},
 		subscribeOutput:      &sns.SubscribeOutput{},
 		subscribeOutputError: nil,
 	}
 
-	err = a.attempSubscription(0)
+	err = a.reconcileSubscription()
 	assert.NoError(t, err)
 }
 
diff --git a/pkg/adapter/awssnssource/testSNSConfirmSubscriptionEvent.json b/pkg/adapter/awssnssource/testSNSConfirmSubscriptionEvent.json
index a0a2337..8433002 100644
--- a/pkg/adapter/awssnssource/testSNSConfirmSubscriptionEvent.json
+++ b/pkg/adapter/awssnssource/testSNSConfirmSubscriptionEvent.json
@@ -1,12 +1,12 @@
 {
   "Type" : "SubscriptionConfirmation",
-  "MessageId" : "6d80ae91-b993-4bd8-84c8-fc089210b102",
-  "Token" : "token",
-  "TopicArn" : "arn:aws:sns:us-west-2:730809894724:triggermeshtest",
-  "Message" : "You have chosen to subscribe to the topic arn:aws:sns:us-west-2:730809894724:triggermeshtest.\nTo confirm the subscription, visit the SubscribeURL included in this message.",
-  "SubscribeURL" : "https://example.com",
-  "Timestamp" : "2019-02-16T07:55:09.097Z",
+  "MessageId" : "165545c9-2a5c-472c-8df2-7ff2be2b3b1b",
+  "Token" : "2336412f37...",
+  "TopicArn" : "arn:aws:sns:us-west-2:123456789012:MyTopic",
+  "Message" : "You have chosen to subscribe to the topic arn:aws:sns:us-west-2:123456789012:MyTopic.\nTo confirm the subscription, visit the SubscribeURL included in this message.",
+  "SubscribeURL" : "https://sns.us-west-2.amazonaws.com/?Action=ConfirmSubscription&TopicArn=arn:aws:sns:us-west-2:123456789012:MyTopic&Token=2336412f37...",
+  "Timestamp" : "2012-04-26T20:45:04.751Z",
   "SignatureVersion" : "1",
-  "Signature" : "2YTrQ==",
-  "SigningCertURL" : "https://sns.us-west-2.amazonaws.com/SimpleNotificationService-ac.pem"
+  "Signature" : "EXAMPLEpH+DcEwjAPg8O9mY8dReBSwksfg2S7WKQcikcNKWLQjwu6A4VbeS0QHVCkhRS7fUQvi2egU3N858fiTDN6bkkOxYDVrY0Ad8L10Hs3zH81mtnPk5uvvolIC1CXGu43obcgFxeL3khZl8IKvO61GWB6jI9b5+gLPoBc1Q=",
+  "SigningCertURL" : "https://sns.us-west-2.amazonaws.com/SimpleNotificationService-f3ecfb7224c7233fe7bb5f59f96de52f.pem"
 }
diff --git a/pkg/adapter/awssnssource/testSNSNotificationEvent.json b/pkg/adapter/awssnssource/testSNSNotificationEvent.json
index 5b551c0..6dc509d 100644
--- a/pkg/adapter/awssnssource/testSNSNotificationEvent.json
+++ b/pkg/adapter/awssnssource/testSNSNotificationEvent.json
@@ -1,15 +1,12 @@
 {
   "Type" : "Notification",
-  "MessageId" : "c2f44639-64fc-50dc-8b9e-12d8a0c39dc7",
-  "TopicArn" : "arn:aws:sns:us-west-2:730809894724:triggermeshtest",
-  "Subject" : "Hello World",
-  "Message" : "fsdklfjskljskldjf",
-  "Timestamp" : "2019-02-16T07:55:43.971Z",
+  "MessageId" : "22b80b92-fdea-4c2c-8f9d-bdfb0c7bf324",
+  "TopicArn" : "arn:aws:sns:us-west-2:123456789012:MyTopic",
+  "Subject" : "My First Message",
+  "Message" : "Hello world!",
+  "Timestamp" : "2012-05-02T00:54:06.655Z",
   "SignatureVersion" : "1",
-  "Signature" : "HY==",
-  "SigningCertURL" : "https://sns.us-west-2.amazonaws.com/SimpleNotificationService-a.pem",
-  "UnsubscribeURL" : "https://sns.us-west-2.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-west-2",
-  "MessageAttributes" : {
-    "bar" : {"Type":"String","Value":"Foo"}
-  }
+  "Signature" : "EXAMPLEw6JRN...",
+  "SigningCertURL" : "https://sns.us-west-2.amazonaws.com/SimpleNotificationService-f3ecfb7224c7233fe7bb5f59f96de52f.pem",
+  "UnsubscribeURL" : "https://sns.us-west-2.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-west-2:123456789012:MyTopic:c9135db0-26c4-47ec-8998-413945fb5a96"
 }
diff --git a/pkg/adapter/common/common_test.go b/pkg/adapter/common/common_test.go
index 9a2b1fe..fe1a9b3 100644
--- a/pkg/adapter/common/common_test.go
+++ b/pkg/adapter/common/common_test.go
@@ -120,7 +120,7 @@ func TestMustParseResource(t *testing.T) {
 	}
 }
 
-func TestMustParseCognitoResource(t *testing.T) {
+func TestMustParseCognitoIdentityResource(t *testing.T) {
 	testCases := map[string]struct {
 		input       string
 		expect      string
@@ -141,12 +141,48 @@ func TestMustParseCognitoResource(t *testing.T) {
 		t.Run(name, func(t *testing.T) {
 			var out string
 			var testFn assert.PanicTestFunc = func() {
-				out = MustParseCognitoResource(tc.input)
+				out = MustParseCognitoIdentityResource(tc.input)
 			}
 
 			if tc.expectPanic {
 				assert.PanicsWithError(t,
-					newParseResourceError(expectCognitoResourceFmt, tc.input).Error(), testFn)
+					newParseResourceError(expectCognitoIdentityResourceFmt, tc.input).Error(), testFn)
+			} else {
+				assert.NotPanics(t, testFn)
+			}
+
+			assert.Equal(t, tc.expect, out)
+		})
+	}
+}
+
+func TestMustParseCognitoUserPoolResource(t *testing.T) {
+	testCases := map[string]struct {
+		input       string
+		expect      string
+		expectPanic bool
+	}{
+		"valid input": {
+			input:  "userpool/some-value",
+			expect: "some-value",
+		},
+		"invalid input": {
+			input:       "not-userpool/some-value",
+			expectPanic: true,
+		},
+	}
+
+	for name, tc := range testCases {
+		//nolint:scopelint
+		t.Run(name, func(t *testing.T) {
+			var out string
+			var testFn assert.PanicTestFunc = func() {
+				out = MustParseCognitoUserPoolResource(tc.input)
+			}
+
+			if tc.expectPanic {
+				assert.PanicsWithError(t,
+					newParseResourceError(expectCognitoUserPoolResourceFmt, tc.input).Error(), testFn)
 			} else {
 				assert.NotPanics(t, testFn)
 			}
diff --git a/pkg/reconciler/awscognitosource/controller_test.go b/pkg/reconciler/awscognitoidentitysource/controller_test.go
similarity index 92%
rename from pkg/reconciler/awscognitosource/controller_test.go
rename to pkg/reconciler/awscognitoidentitysource/controller_test.go
index b0ca788..df55393 100644
--- a/pkg/reconciler/awscognitosource/controller_test.go
+++ b/pkg/reconciler/awscognitoidentitysource/controller_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	"testing"
@@ -22,7 +22,7 @@ import (
 	. "github.com/triggermesh/aws-event-sources/pkg/reconciler/testing"
 
 	// Link fake informers accessed by our controller
-	_ "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitosource/fake"
+	_ "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/informers/sources/v1alpha1/awscognitoidentitysource/fake"
 	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment/fake"
 	_ "knative.dev/pkg/injection/clients/dynamicclient/fake"
diff --git a/pkg/reconciler/awscognitosource/reconciler_test.go b/pkg/reconciler/awscognitoidentitysource/reconciler_test.go
similarity index 89%
rename from pkg/reconciler/awscognitosource/reconciler_test.go
rename to pkg/reconciler/awscognitoidentitysource/reconciler_test.go
index d2500af..ec76f87 100644
--- a/pkg/reconciler/awscognitosource/reconciler_test.go
+++ b/pkg/reconciler/awscognitoidentitysource/reconciler_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package awscognitosource
+package awscognitoidentitysource
 
 import (
 	"context"
@@ -33,7 +33,7 @@ import (
 
 	"github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
 	fakeinjectionclient "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/client/fake"
-	reconcilerv1alpha1 "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitosource"
+	reconcilerv1alpha1 "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/reconciler/sources/v1alpha1/awscognitoidentitysource"
 	"github.com/triggermesh/aws-event-sources/pkg/reconciler/common"
 	. "github.com/triggermesh/aws-event-sources/pkg/reconciler/testing"
 )
@@ -53,7 +53,7 @@ func TestReconcileSource(t *testing.T) {
 	TestReconcile(t, ctor, src, adapterFn)
 }
 
-// reconcilerCtor returns a Ctor for a AWSCognitoSource Reconciler.
+// reconcilerCtor returns a Ctor for a AWSCognitoIdentitySource Reconciler.
 func reconcilerCtor(cfg *adapterConfig) Ctor {
 	return func(t *testing.T, ctx context.Context, ls *Listers) controller.Reconciler {
 		base := common.GenericDeploymentReconciler{
@@ -68,15 +68,15 @@ func reconcilerCtor(cfg *adapterConfig) Ctor {
 		}
 
 		return reconcilerv1alpha1.NewReconciler(ctx, logging.FromContext(ctx),
-			fakeinjectionclient.Get(ctx), ls.GetAWSCognitoSourceLister(),
+			fakeinjectionclient.Get(ctx), ls.GetAWSCognitoIdentitySourceLister(),
 			controller.GetEventRecorder(ctx), r)
 	}
 }
 
 // newEventSource returns a test source object with a minimal set of pre-filled attributes.
-func newEventSource() *v1alpha1.AWSCognitoSource {
-	src := &v1alpha1.AWSCognitoSource{
-		Spec: v1alpha1.AWSCognitoSourceSpec{
+func newEventSource() *v1alpha1.AWSCognitoIdentitySource {
+	src := &v1alpha1.AWSCognitoIdentitySource{
+		Spec: v1alpha1.AWSCognitoIdentitySourceSpec{
 			ARN: NewARN(cognitoidentity.ServiceName, "identitypool/triggermeshtest").String(),
 			Credentials: v1alpha1.AWSSecurityCredentials{
 				AccessKeyID: v1alpha1.ValueFromField{
diff --git a/pkg/reconciler/awscognitouserpoolsource/reconciler_test.go b/pkg/reconciler/awscognitouserpoolsource/reconciler_test.go
index 2a04858..65a8cdd 100644
--- a/pkg/reconciler/awscognitouserpoolsource/reconciler_test.go
+++ b/pkg/reconciler/awscognitouserpoolsource/reconciler_test.go
@@ -20,7 +20,7 @@ import (
 	"context"
 	"testing"
 
-	"github.com/aws/aws-sdk-go/service/cognitoidentity"
+	"github.com/aws/aws-sdk-go/service/cognitoidentityprovider"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -77,7 +77,7 @@ func reconcilerCtor(cfg *adapterConfig) Ctor {
 func newEventSource(skipCEAtrributes ...interface{}) *v1alpha1.AWSCognitoUserPoolSource {
 	src := &v1alpha1.AWSCognitoUserPoolSource{
 		Spec: v1alpha1.AWSCognitoUserPoolSourceSpec{
-			ARN: NewARN(cognitoidentity.ServiceName, "userpool/triggermeshtest").String(),
+			ARN: NewARN(cognitoidentityprovider.ServiceName, "userpool/triggermeshtest").String(),
 			Credentials: v1alpha1.AWSSecurityCredentials{
 				AccessKeyID: v1alpha1.ValueFromField{
 					ValueFromSecret: &corev1.SecretKeySelector{
diff --git a/pkg/reconciler/awssnssource/controller_test.go b/pkg/reconciler/awssnssource/controller_test.go
index e4c555f..21b6825 100644
--- a/pkg/reconciler/awssnssource/controller_test.go
+++ b/pkg/reconciler/awssnssource/controller_test.go
@@ -24,8 +24,8 @@ import (
 	// Link fake informers accessed by our controller
 	_ "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/informers/sources/v1alpha1/awssnssource/fake"
 	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
-	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment/fake"
 	_ "knative.dev/pkg/injection/clients/dynamicclient/fake"
+	_ "knative.dev/serving/pkg/client/injection/informers/serving/v1/service/fake"
 )
 
 func TestNewController(t *testing.T) {
diff --git a/pkg/reconciler/awssnssource/reconciler_test.go b/pkg/reconciler/awssnssource/reconciler_test.go
index 434159e..570317a 100644
--- a/pkg/reconciler/awssnssource/reconciler_test.go
+++ b/pkg/reconciler/awssnssource/reconciler_test.go
@@ -20,16 +20,17 @@ import (
 	"context"
 	"testing"
 
+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/sns"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 
 	"knative.dev/eventing/pkg/reconciler/source"
-	fakek8sinjectionclient "knative.dev/pkg/client/injection/kube/client/fake"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/resolver"
+	fakeservinginjectionclient "knative.dev/serving/pkg/client/injection/client/fake"
 
 	"github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
 	fakeinjectionclient "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/client/fake"
@@ -47,7 +48,7 @@ func TestReconcileSource(t *testing.T) {
 	var (
 		ctor      = reconcilerCtor(adapterCfg)
 		src       = newEventSource()
-		adapterFn = adapterDeploymentBuilder(src, adapterCfg)
+		adapterFn = adapterServiceBuilder(src, adapterCfg)
 	)
 
 	TestReconcile(t, ctor, src, adapterFn)
@@ -56,10 +57,10 @@ func TestReconcileSource(t *testing.T) {
 // reconcilerCtor returns a Ctor for a AWSSNSSource Reconciler.
 func reconcilerCtor(cfg *adapterConfig) Ctor {
 	return func(t *testing.T, ctx context.Context, ls *Listers) controller.Reconciler {
-		base := common.GenericDeploymentReconciler{
+		base := common.GenericServiceReconciler{
 			SinkResolver: resolver.NewURIResolver(ctx, func(types.NamespacedName) {}),
-			Lister:       ls.GetDeploymentLister().Deployments,
-			Client:       fakek8sinjectionclient.Get(ctx).AppsV1().Deployments,
+			Lister:       ls.GetServiceLister().Services,
+			Client:       fakeservinginjectionclient.Get(ctx).ServingV1().Services,
 		}
 
 		r := &Reconciler{
@@ -78,6 +79,9 @@ func newEventSource(skipCEAtrributes ...interface{}) *v1alpha1.AWSSNSSource {
 	src := &v1alpha1.AWSSNSSource{
 		Spec: v1alpha1.AWSSNSSourceSpec{
 			ARN: NewARN(sns.ServiceName, "triggermeshtest").String(),
+			SubscriptionAttributes: map[string]*string{
+				"DeliveryPolicy": aws.String(`{"healthyRetryPolicy":{"numRetries":5}}`),
+			},
 			Credentials: v1alpha1.AWSSecurityCredentials{
 				AccessKeyID: v1alpha1.ValueFromField{
 					ValueFromSecret: &corev1.SecretKeySelector{
diff --git a/pkg/reconciler/common/resource/knservice_test.go b/pkg/reconciler/common/resource/knservice_test.go
new file mode 100644
index 0000000..fbadd96
--- /dev/null
+++ b/pkg/reconciler/common/resource/knservice_test.go
@@ -0,0 +1,133 @@
+/*
+Copyright (c) 2020 TriggerMesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package resource
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	servingv1 "knative.dev/serving/pkg/apis/serving/v1"
+)
+
+func TestNewServiceWithDefaultContainer(t *testing.T) {
+	ksvc := NewKnService(tNs, tName,
+		PodLabel("test.podlabel/2", "val2"),
+		Port("health", 8081),
+		Image(tImg),
+		PodLabel("test.podlabel/1", "val1"),
+		EnvVar("TEST_ENV1", "val1"),
+		Port("h2c", 8080), // overrides previously defined port
+		Label("test.label/1", "val1"),
+		Probe("/health", "health"), // port is ignored
+		EnvVars(makeEnvVars(2, "MULTI_ENV", "val")...),
+		EnvVar("TEST_ENV2", "val2"),
+		Label("test.label/2", "val2"),
+	)
+
+	expectKsvc := &servingv1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: tNs,
+			Name:      tName,
+			Labels: map[string]string{
+				"test.label/1": "val1",
+				"test.label/2": "val2",
+			},
+		},
+		Spec: servingv1.ServiceSpec{
+			ConfigurationSpec: servingv1.ConfigurationSpec{
+				Template: servingv1.RevisionTemplateSpec{
+					ObjectMeta: metav1.ObjectMeta{
+						Labels: map[string]string{
+							"test.podlabel/1": "val1",
+							"test.podlabel/2": "val2",
+						},
+					},
+					Spec: servingv1.RevisionSpec{
+						PodSpec: corev1.PodSpec{
+							Containers: []corev1.Container{{
+								Name:  defaultContainerName,
+								Image: tImg,
+								Ports: []corev1.ContainerPort{{
+									Name:          "h2c",
+									ContainerPort: 8080,
+								}},
+								Env: []corev1.EnvVar{{
+									Name:  "TEST_ENV1",
+									Value: "val1",
+								}, {
+									Name:  "MULTI_ENV1",
+									Value: "val1",
+								}, {
+									Name:  "MULTI_ENV2",
+									Value: "val2",
+								}, {
+									Name:  "TEST_ENV2",
+									Value: "val2",
+								}},
+								ReadinessProbe: &corev1.Probe{
+									Handler: corev1.Handler{
+										HTTPGet: &corev1.HTTPGetAction{
+											Path: "/health",
+										},
+									},
+								},
+							}},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	if d := cmp.Diff(expectKsvc, ksvc); d != "" {
+		t.Errorf("Unexpected diff: (-:expect, +:got) %s", d)
+	}
+}
+
+func TestNewServiceWithCustomContainer(t *testing.T) {
+	ksvc := NewKnService(tNs, tName,
+		Container(&corev1.Container{Name: "foo"}),
+	)
+
+	expectKsvc := &servingv1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: tNs,
+			Name:      tName,
+		},
+		Spec: servingv1.ServiceSpec{
+			ConfigurationSpec: servingv1.ConfigurationSpec{
+				Template: servingv1.RevisionTemplateSpec{
+					Spec: servingv1.RevisionSpec{
+						PodSpec: corev1.PodSpec{
+							Containers: []corev1.Container{{
+								Name: "foo",
+							}},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	if d := cmp.Diff(expectKsvc, ksvc); d != "" {
+		t.Errorf("Unexpected diff: (-:expect, +:got) %s", d)
+	}
+}
diff --git a/pkg/reconciler/common/semantic/semantic_test.go b/pkg/reconciler/common/semantic/semantic_test.go
index b156d8b..db739e3 100644
--- a/pkg/reconciler/common/semantic/semantic_test.go
+++ b/pkg/reconciler/common/semantic/semantic_test.go
@@ -25,9 +25,14 @@ import (
 
 	appsv1 "k8s.io/api/apps/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+
+	servingv1 "knative.dev/serving/pkg/apis/serving/v1"
 )
 
-const fixtureDeploymentPath = "../../../../test/fixtures/deployment.json"
+const (
+	fixtureDeploymentPath = "../../../../test/fixtures/deployment.json"
+	fixtureKnServicePath  = "../../../../test/fixtures/knService.json"
+)
 
 func TestDeploymentEqual(t *testing.T) {
 	current := &appsv1.Deployment{}
@@ -104,6 +109,81 @@ func TestDeploymentEqual(t *testing.T) {
 	}
 }
 
+func TestKnServiceEqual(t *testing.T) {
+	current := &servingv1.Service{}
+	loadFixture(t, fixtureKnServicePath, current)
+
+	assert.GreaterOrEqual(t, len(current.Labels), 2,
+		"Test suite requires a reference object with at least 2 labels to run properly")
+
+	assert.True(t, knServiceEqual(nil, nil), "Two nil elements should be equal")
+
+	testCases := map[string]struct {
+		prep   func() *servingv1.Service
+		expect bool
+	}{
+		"not equal when one element is nil": {
+			func() *servingv1.Service {
+				return nil
+			},
+			false,
+		},
+		// counter intuitive but expected result for deep derivative comparisons
+		"equal when all desired attributes are empty": {
+			func() *servingv1.Service {
+				return &servingv1.Service{}
+			},
+			true,
+		},
+		"not equal when some existing attribute differs": {
+			func() *servingv1.Service {
+				desired := current.DeepCopy()
+				for k := range desired.Labels {
+					desired.Labels[k] += "test"
+					break // changing one is enough
+				}
+				return desired
+			},
+			false,
+		},
+		"equal when current has more attributes than desired": {
+			func() *servingv1.Service {
+				desired := current.DeepCopy()
+				for k := range desired.Labels {
+					delete(desired.Labels, k)
+					break // deleting one is enough
+				}
+				return desired
+			},
+			true,
+		},
+		"not equal when desired has more attributes than current": {
+			func() *servingv1.Service {
+				desired := current.DeepCopy()
+				for k := range desired.Labels {
+					desired.Labels[k+"test"] = "test"
+					break // adding one is enough
+				}
+				return desired
+			},
+			false,
+		},
+	}
+
+	for name, tc := range testCases {
+		// //nolint:scopelint
+		t.Run(name, func(t *testing.T) {
+			desired := tc.prep()
+			switch tc.expect {
+			case true:
+				assert.True(t, knServiceEqual(desired, current))
+			case false:
+				assert.False(t, knServiceEqual(desired, current))
+			}
+		})
+	}
+}
+
 func loadFixture(t *testing.T, file string, obj runtime.Object) {
 	t.Helper()
 
diff --git a/pkg/reconciler/testing/factory.go b/pkg/reconciler/testing/factory.go
index d67ff72..28198dc 100644
--- a/pkg/reconciler/testing/factory.go
+++ b/pkg/reconciler/testing/factory.go
@@ -31,6 +31,7 @@ import (
 	"knative.dev/pkg/logging"
 	logtesting "knative.dev/pkg/logging/testing"
 	rt "knative.dev/pkg/reconciler/testing"
+	fakeservinginjectionclient "knative.dev/serving/pkg/client/injection/client/fake"
 
 	fakeinjectionclient "github.com/triggermesh/aws-event-sources/pkg/client/generated/injection/client/fake"
 )
@@ -60,6 +61,7 @@ func MakeFactory(ctor Ctor) rt.Factory {
 		// all clients used inside reconciler implementations should be
 		// injected as well
 		ctx, k8sClient := fakek8sinjectionclient.With(ctx, ls.GetKubeObjects()...)
+		ctx, servingClient := fakeservinginjectionclient.With(ctx, ls.GetServingObjects()...)
 
 		// duck informers (e.g. used by resolver.URIResolver) use a dynamic client.
 		ctx, _ = fakedynamicclient.With(ctx, scheme, ToUnstructured(t, tr.Objects)...)
@@ -80,11 +82,13 @@ func MakeFactory(ctor Ctor) rt.Factory {
 		for _, reactor := range tr.WithReactors {
 			client.PrependReactor("*", "*", reactor)
 			k8sClient.PrependReactor("*", "*", reactor)
+			servingClient.PrependReactor("*", "*", reactor)
 		}
 
 		actionRecorderList := rt.ActionRecorderList{
 			client, // record status updates
 			k8sClient,
+			servingClient,
 		}
 
 		eventList := rt.EventList{
diff --git a/pkg/reconciler/testing/listers.go b/pkg/reconciler/testing/listers.go
index 08974f8..ba62064 100644
--- a/pkg/reconciler/testing/listers.go
+++ b/pkg/reconciler/testing/listers.go
@@ -27,6 +27,9 @@ import (
 
 	fakeeventingclientset "knative.dev/eventing/pkg/client/clientset/versioned/fake"
 	rt "knative.dev/pkg/reconciler/testing"
+	servingv1 "knative.dev/serving/pkg/apis/serving/v1"
+	fakeservingclient "knative.dev/serving/pkg/client/clientset/versioned/fake"
+	servinglistersv1 "knative.dev/serving/pkg/client/listers/serving/v1"
 
 	"github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
 	fakeclient "github.com/triggermesh/aws-event-sources/pkg/client/generated/clientset/internalclientset/fake"
@@ -36,6 +39,7 @@ import (
 var clientSetSchemes = []func(*runtime.Scheme) error{
 	fakeclient.AddToScheme,
 	fakek8sclient.AddToScheme,
+	fakeservingclient.AddToScheme,
 	// although our reconcilers do not handle eventing objects directly, we
 	// do need to register the eventing Scheme so that sink URI resolvers
 	// can recongnize the Broker objects we use in tests
@@ -85,21 +89,26 @@ func (l *Listers) GetKubeObjects() []runtime.Object {
 	return l.sorter.ObjectsForSchemeFunc(fakek8sclient.AddToScheme)
 }
 
+// GetServingObjects returns objects from the serving API.
+func (l *Listers) GetServingObjects() []runtime.Object {
+	return l.sorter.ObjectsForSchemeFunc(fakeservingclient.AddToScheme)
+}
+
 // GetAWSCodeCommitSourceLister returns a Lister for AWSCodeCommitSource objects.
 func (l *Listers) GetAWSCodeCommitSourceLister() listersv1alpha1.AWSCodeCommitSourceLister {
 	return listersv1alpha1.NewAWSCodeCommitSourceLister(l.IndexerFor(&v1alpha1.AWSCodeCommitSource{}))
 }
 
-// GetAWSCognitoSourceLister returns a Lister for AWSCognitoSource objects.
-func (l *Listers) GetAWSCognitoSourceLister() listersv1alpha1.AWSCognitoSourceLister {
-	return listersv1alpha1.NewAWSCognitoSourceLister(l.IndexerFor(&v1alpha1.AWSCognitoSource{}))
-}
-
-// GetAWSCognitoUserPoolSourceLister returns a Lister for AWSCognitoUserPoolSource objects.
+// GetAWSCognitoUserPoolSourceLister returns a Lister for AWSCognitoIdentitySource objects.
 func (l *Listers) GetAWSCognitoUserPoolSourceLister() listersv1alpha1.AWSCognitoUserPoolSourceLister {
 	return listersv1alpha1.NewAWSCognitoUserPoolSourceLister(l.IndexerFor(&v1alpha1.AWSCognitoUserPoolSource{}))
 }
 
+// GetAWSCognitoIdentitySourceLister returns a Lister for AWSCognitoUserPoolSource objects.
+func (l *Listers) GetAWSCognitoIdentitySourceLister() listersv1alpha1.AWSCognitoIdentitySourceLister {
+	return listersv1alpha1.NewAWSCognitoIdentitySourceLister(l.IndexerFor(&v1alpha1.AWSCognitoIdentitySource{}))
+}
+
 // GetAWSDynamoDBSourceLister returns a Lister for AWSDynamoDBSource objects.
 func (l *Listers) GetAWSDynamoDBSourceLister() listersv1alpha1.AWSDynamoDBSourceLister {
 	return listersv1alpha1.NewAWSDynamoDBSourceLister(l.IndexerFor(&v1alpha1.AWSDynamoDBSource{}))
@@ -129,3 +138,8 @@ func (l *Listers) GetAWSSQSSourceLister() listersv1alpha1.AWSSQSSourceLister {
 func (l *Listers) GetDeploymentLister() k8slistersv1.DeploymentLister {
 	return k8slistersv1.NewDeploymentLister(l.IndexerFor(&appsv1.Deployment{}))
 }
+
+// GetServiceLister returns a lister for Service objects.
+func (l *Listers) GetServiceLister() servinglistersv1.ServiceLister {
+	return servinglistersv1.NewServiceLister(l.IndexerFor(&servingv1.Service{}))
+}
diff --git a/pkg/reconciler/testing/reconcile.go b/pkg/reconciler/testing/reconcile.go
index a25ee82..92a2694 100644
--- a/pkg/reconciler/testing/reconcile.go
+++ b/pkg/reconciler/testing/reconcile.go
@@ -23,6 +23,7 @@ import (
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -33,6 +34,7 @@ import (
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	rt "knative.dev/pkg/reconciler/testing"
+	servingv1 "knative.dev/serving/pkg/apis/serving/v1"
 
 	"github.com/aws/aws-sdk-go/aws/arn"
 
@@ -61,15 +63,14 @@ var tSinkURI = &apis.URL{
 //  2. MakeFactory injects those clients into a context along with fake event recorders, etc.
 //  3. A Reconciler is constructed via a Ctor function using the values injected above
 //  4. The Reconciler returned by MakeFactory is used to run the test case
-func TestReconcile(t *testing.T, ctor Ctor, src v1alpha1.AWSEventSource,
-	adapterFn common.AdapterDeploymentBuilderFunc) {
-
+func TestReconcile(t *testing.T, ctor Ctor, src v1alpha1.AWSEventSource, adapterFn interface{}) {
 	assertPopulatedSource(t, src)
 
 	newEventSource := eventSourceCtor(src)
-	newAdapterDeployment := adapterDeploymentCtor(adapterFn, src)
+	newAdapter := adapterCtor(adapterFn, src)
 
-	n := newAdapterDeployment().Name
+	a := newAdapter()
+	n, k, r := nameKindAndResource(a)
 
 	testCases := rt.TableTest{
 		// Creation/Deletion
@@ -82,13 +83,13 @@ func TestReconcile(t *testing.T, ctor Ctor, src v1alpha1.AWSEventSource,
 				newEventSource(noCEAttributes),
 			},
 			WantCreates: []runtime.Object{
-				newAdapterDeployment(),
+				newAdapter(),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: newEventSource(withSink, notDeployed),
+				Object: newEventSource(withSink, notDeployed(a)),
 			}},
 			WantEvents: []string{
-				createAdapterEvent(n),
+				createAdapterEvent(n, k),
 			},
 		},
 		{
@@ -106,11 +107,11 @@ func TestReconcile(t *testing.T, ctor Ctor, src v1alpha1.AWSEventSource,
 			Key:  tKey,
 			Objects: []runtime.Object{
 				newAdressable(),
-				newEventSource(withSink, notDeployed),
-				newAdapterDeployment(ready),
+				newEventSource(withSink, notDeployed(a)),
+				newAdapter(ready),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: newEventSource(withSink, deployed),
+				Object: newEventSource(withSink, deployed(a)),
 			}},
 		},
 		{
@@ -118,11 +119,11 @@ func TestReconcile(t *testing.T, ctor Ctor, src v1alpha1.AWSEventSource,
 			Key:  tKey,
 			Objects: []runtime.Object{
 				newAdressable(),
-				newEventSource(withSink, deployed),
-				newAdapterDeployment(notReady),
+				newEventSource(withSink, deployed(a)),
+				newAdapter(notReady),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: newEventSource(withSink, notDeployed),
+				Object: newEventSource(withSink, notDeployed(a)),
 			}},
 		},
 		{
@@ -130,14 +131,14 @@ func TestReconcile(t *testing.T, ctor Ctor, src v1alpha1.AWSEventSource,
 			Key:  tKey,
 			Objects: []runtime.Object{
 				newAdressable(),
-				newEventSource(withSink, deployed),
-				newAdapterDeployment(ready, bumpImage),
+				newEventSource(withSink, deployed(a)),
+				newAdapter(ready, bumpImage),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: newAdapterDeployment(ready),
+				Object: newAdapter(ready),
 			}},
 			WantEvents: []string{
-				updateAdapterEvent(n),
+				updateAdapterEvent(n, k),
 			},
 		},
 
@@ -148,11 +149,11 @@ func TestReconcile(t *testing.T, ctor Ctor, src v1alpha1.AWSEventSource,
 			Key:  tKey,
 			Objects: []runtime.Object{
 				/* sink omitted */
-				newEventSource(withSink, deployed),
-				newAdapterDeployment(ready),
+				newEventSource(withSink, deployed(a)),
+				newAdapter(ready),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: newEventSource(withoutSink, deployed),
+				Object: newEventSource(withoutSink, deployed(a)),
 			}},
 			WantEvents: []string{
 				badSinkEvent(),
@@ -160,42 +161,42 @@ func TestReconcile(t *testing.T, ctor Ctor, src v1alpha1.AWSEventSource,
 			WantErr: true,
 		},
 		{
-			Name: "Fail to create adapter deployment",
+			Name: "Fail to create adapter",
 			Key:  tKey,
 			WithReactors: []clientgotesting.ReactionFunc{
-				rt.InduceFailure("create", "deployments"),
+				rt.InduceFailure("create", r),
 			},
 			Objects: []runtime.Object{
 				newAdressable(),
 				newEventSource(withSink),
 			},
 			WantCreates: []runtime.Object{
-				newAdapterDeployment(),
+				newAdapter(),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: newEventSource(unknownDeployedWithError, withSink),
+				Object: newEventSource(unknownDeployedWithError(a), withSink),
 			}},
 			WantEvents: []string{
-				failCreateAdapterEvent(n),
+				failCreateAdapterEvent(n, k, r),
 			},
 			WantErr: true,
 		},
 		{
-			Name: "Fail to update adapter deployment",
+			Name: "Fail to update adapter",
 			Key:  tKey,
 			WithReactors: []clientgotesting.ReactionFunc{
-				rt.InduceFailure("update", "deployments"),
+				rt.InduceFailure("update", r),
 			},
 			Objects: []runtime.Object{
 				newAdressable(),
-				newEventSource(deployed, withSink),
-				newAdapterDeployment(ready, bumpImage),
+				newEventSource(withSink, deployed(a)),
+				newAdapter(ready, bumpImage),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: newAdapterDeployment(ready),
+				Object: newAdapter(ready),
 			}},
 			WantEvents: []string{
-				failUpdateAdapterEvent(n),
+				failUpdateAdapterEvent(n, k, r),
 			},
 			WantErr: true,
 		},
@@ -231,6 +232,24 @@ func assertPopulatedSource(t *testing.T, src v1alpha1.AWSEventSource) {
 	assert.NotEmpty(t, src.GetStatus().Status.Conditions, "Provided source should have initialized conditions")
 }
 
+func nameKindAndResource(object runtime.Object) (string /*name*/, string /*kind*/, string /*resource*/) {
+	metaObj, _ := meta.Accessor(object)
+	name := metaObj.GetName()
+
+	var kind, resource string
+
+	switch object.(type) {
+	case *appsv1.Deployment:
+		kind = "Deployment"
+		resource = "deployments"
+	case *servingv1.Service:
+		kind = "Service"
+		resource = "services"
+	}
+
+	return name, kind, resource
+}
+
 /* Event sources */
 
 // Populate populates an event source with generic attributes.
@@ -294,18 +313,39 @@ func withoutSink(src v1alpha1.AWSEventSource) {
 }
 
 // Deployed: True
-func deployed(src v1alpha1.AWSEventSource) {
-	src.GetStatus().PropagateAvailability(newReadyDeployment())
+func deployed(adapter runtime.Object) sourceOption {
+	adapter = adapter.DeepCopyObject()
+	ready(adapter)
+
+	return func(src v1alpha1.AWSEventSource) {
+		src.GetStatus().PropagateAvailability(adapter)
+	}
 }
 
 // Deployed: False
-func notDeployed(src v1alpha1.AWSEventSource) {
-	src.GetStatus().PropagateAvailability(newNotReadyDeployment())
+func notDeployed(adapter runtime.Object) sourceOption {
+	adapter = adapter.DeepCopyObject()
+	notReady(adapter)
+
+	return func(src v1alpha1.AWSEventSource) {
+		src.GetStatus().PropagateAvailability(adapter)
+	}
 }
 
 // Deployed: Unknown with error
-func unknownDeployedWithError(src v1alpha1.AWSEventSource) {
-	src.GetStatus().PropagateAvailability(nil)
+func unknownDeployedWithError(adapter runtime.Object) sourceOption {
+	var nilObj runtime.Object
+
+	switch adapter.(type) {
+	case *appsv1.Deployment:
+		nilObj = (*appsv1.Deployment)(nil)
+	case *servingv1.Service:
+		nilObj = (*servingv1.Service)(nil)
+	}
+
+	return func(src v1alpha1.AWSEventSource) {
+		src.GetStatus().PropagateAvailability(nilObj)
+	}
 }
 
 // deleted marks the source as deleted.
@@ -314,69 +354,83 @@ func deleted(src v1alpha1.AWSEventSource) {
 	src.SetDeletionTimestamp(&t)
 }
 
-/* Adapter deployment */
+/* Adapter */
 
-// adapterCtorWithOptions is a function that returns a Deployment object with
+// adapterCtorWithOptions is a function that returns a runtime object with
 // modifications applied.
-type adapterCtorWithOptions func(...deploymentOption) *appsv1.Deployment
-
-// adapterDeploymentCtor creates a copy of the given Deployment and returns a
-// function that can apply options to that Deployment.
-func adapterDeploymentCtor(adapterFn common.AdapterDeploymentBuilderFunc,
-	src v1alpha1.AWSEventSource) adapterCtorWithOptions {
+type adapterCtorWithOptions func(...adapterOption) runtime.Object
 
+// adapterCtor creates a copy of the given adapter object and returns a
+// function that can apply options to that object.
+func adapterCtor(adapterFn interface{}, src v1alpha1.AWSEventSource) adapterCtorWithOptions {
 	// error discarded because the ARN format is already validated during tests
 	arn, _ := arn.Parse(src.GetARN())
 
-	return func(opts ...deploymentOption) *appsv1.Deployment {
-		d := adapterFn(arn, tSinkURI)
+	return func(opts ...adapterOption) runtime.Object {
+		var obj runtime.Object
+
+		switch typedAdapterFn := adapterFn.(type) {
+		case common.AdapterDeploymentBuilderFunc:
+			obj = typedAdapterFn(arn, tSinkURI)
+		case common.AdapterServiceBuilderFunc:
+			obj = typedAdapterFn(arn, tSinkURI)
+		}
 
 		for _, opt := range opts {
-			opt(d)
+			opt(obj)
 		}
 
-		return d
+		return obj
 	}
 }
 
-// deploymentOption is a functional option for a Deployment object.
-type deploymentOption func(*appsv1.Deployment)
+// adapterOption is a functional option for an adapter object.
+type adapterOption func(runtime.Object)
 
 // Ready: True
-func ready(d *appsv1.Deployment) {
-	d.Status = appsv1.DeploymentStatus{
-		Conditions: []appsv1.DeploymentCondition{{
-			Type:   appsv1.DeploymentAvailable,
+func ready(object runtime.Object) {
+	switch o := object.(type) {
+	case *appsv1.Deployment:
+		o.Status = appsv1.DeploymentStatus{
+			Conditions: []appsv1.DeploymentCondition{{
+				Type:   appsv1.DeploymentAvailable,
+				Status: corev1.ConditionTrue,
+			}},
+		}
+	case *servingv1.Service:
+		o.Status.SetConditions(apis.Conditions{{
+			Type:   v1alpha1.ConditionReady,
 			Status: corev1.ConditionTrue,
-		}},
+		}})
 	}
 }
 
-func newReadyDeployment() *appsv1.Deployment {
-	d := &appsv1.Deployment{}
-	ready(d)
-	return d
-}
-
 // Ready: False
-func notReady(d *appsv1.Deployment) {
-	d.Status = appsv1.DeploymentStatus{
-		Conditions: []appsv1.DeploymentCondition{{
-			Type:   appsv1.DeploymentAvailable,
+func notReady(object runtime.Object) {
+	switch o := object.(type) {
+	case *appsv1.Deployment:
+		o.Status = appsv1.DeploymentStatus{
+			Conditions: []appsv1.DeploymentCondition{{
+				Type:   appsv1.DeploymentAvailable,
+				Status: corev1.ConditionFalse,
+			}},
+		}
+	case *servingv1.Service:
+		o.Status.SetConditions(apis.Conditions{{
+			Type:   v1alpha1.ConditionReady,
 			Status: corev1.ConditionFalse,
-		}},
+		}})
 	}
 }
 
-func newNotReadyDeployment() *appsv1.Deployment {
-	d := &appsv1.Deployment{}
-	notReady(d)
-	return d
-}
-
 // bumpImage adds a static suffix to the Deployment's image.
-func bumpImage(d *appsv1.Deployment) {
-	d.Spec.Template.Spec.Containers[0].Image += "-test"
+func bumpImage(object runtime.Object) {
+	switch o := object.(type) {
+	case *appsv1.Deployment:
+		o.Spec.Template.Spec.Containers[0].Image += "-test"
+	case *servingv1.Service:
+		o.Spec.Template.Spec.Containers[0].Image += "-test"
+	}
 }
 
 /* Event sink */
@@ -398,19 +452,19 @@ func newAdressable() *eventingv1beta1.Broker {
 
 /* Events */
 
-func createAdapterEvent(name string) string {
-	return eventtesting.Eventf(corev1.EventTypeNormal, common.ReasonAdapterCreate, "Created adapter Deployment %q", name)
+func createAdapterEvent(name, kind string) string {
+	return eventtesting.Eventf(corev1.EventTypeNormal, common.ReasonAdapterCreate, "Created adapter %s %q", kind, name)
 }
-func updateAdapterEvent(name string) string {
-	return eventtesting.Eventf(corev1.EventTypeNormal, common.ReasonAdapterUpdate, "Updated adapter Deployment %q", name)
+func updateAdapterEvent(name, kind string) string {
+	return eventtesting.Eventf(corev1.EventTypeNormal, common.ReasonAdapterUpdate, "Updated adapter %s %q", kind, name)
 }
-func failCreateAdapterEvent(name string) string {
-	return eventtesting.Eventf(corev1.EventTypeWarning, common.ReasonFailedAdapterCreate, "Failed to create adapter Deployment %q: "+
-		"inducing failure for create deployments", name)
+func failCreateAdapterEvent(name, kind, resource string) string {
+	return eventtesting.Eventf(corev1.EventTypeWarning, common.ReasonFailedAdapterCreate, "Failed to create adapter %s %q: "+
+		"inducing failure for create %s", kind, name, resource)
 }
-func failUpdateAdapterEvent(name string) string {
-	return eventtesting.Eventf(corev1.EventTypeWarning, common.ReasonFailedAdapterUpdate, "Failed to update adapter Deployment %q: "+
-		"inducing failure for update deployments", name)
+func failUpdateAdapterEvent(name, kind, resource string) string {
+	return eventtesting.Eventf(corev1.EventTypeWarning, common.ReasonFailedAdapterUpdate, "Failed to update adapter %s %q: "+
+		"inducing failure for update %s", kind, name, resource)
 }
 func badSinkEvent() string {
 	addrGVK := newAdressable().GetGroupVersionKind()
diff --git a/test/fixtures/knService.json b/test/fixtures/knService.json
new file mode 100644
index 0000000..f9e897c
--- /dev/null
+++ b/test/fixtures/knService.json
@@ -0,0 +1,109 @@
+{
+    "apiVersion": "serving.knative.dev/v1",
+    "kind": "Service",
+    "metadata": {
+        "annotations": {
+            "serving.knative.dev/creator": "docker-for-desktop",
+            "serving.knative.dev/lastModifier": "docker-for-desktop"
+        },
+        "creationTimestamp": "2020-04-23T21:14:01Z",
+        "generation": 1,
+        "labels": {
+            "app.kubernetes.io/component": "adapter",
+            "app.kubernetes.io/instance": "sample",
+            "app.kubernetes.io/managed-by": "aws-event-sources-controller",
+            "app.kubernetes.io/name": "awssnssource",
+            "app.kubernetes.io/part-of": "aws-event-sources"
+        },
+        "name": "awssnssource-sample",
+        "namespace": "dev",
+        "ownerReferences": [
+            {
+                "apiVersion": "sources.triggermesh.io/v1alpha1",
+                "blockOwnerDeletion": true,
+                "controller": true,
+                "kind": "AWSSNSSource",
+                "name": "sample",
+                "uid": "429cbd19-06ef-4795-b78e-ffcbc30660a1"
+            }
+        ],
+        "resourceVersion": "572071",
+        "selfLink": "/apis/serving.knative.dev/v1/namespaces/dev/services/awssnssource-sample",
+        "uid": "5470339e-86a7-46fe-bb66-4c49c4a8f22e"
+    },
+    "spec": {
+        "template": {
+            "metadata": {
+                "creationTimestamp": null,
+                "labels": {
+                    "app.kubernetes.io/component": "adapter",
+                    "app.kubernetes.io/instance": "sample",
+                    "app.kubernetes.io/managed-by": "aws-event-sources-controller",
+                    "app.kubernetes.io/name": "awssnssource",
+                    "app.kubernetes.io/part-of": "aws-event-sources"
+                }
+            },
+            "spec": {
+                "containerConcurrency": 0,
+                "containers": [
+                    {
+                        "env": [
+                            {
+                                "name": "K_LOGGING_CONFIG",
+                                "value": "{\"zap-logger-config\":\"{\\n  \\\"level\\\": \\\"info\\\",\\n  \\\"development\\\": false,\\n  \\\"outputPaths\\\": [\\\"stdout\\\"],\\n  \\\"errorOutputPaths\\\": [\\\"stderr\\\"],\\n  \\\"encoding\\\": \\\"json\\\",\\n  \\\"encoderConfig\\\": {\\n    \\\"timeKey\\\": \\\"ts\\\",\\n    \\\"levelKey\\\": \\\"level\\\",\\n    \\\"nameKey\\\": \\\"logger\\\",\\n    \\\"callerKey\\\": \\\"caller\\\",\\n    \\\"messageKey\\\": \\\"msg\\\",\\n    \\\"stacktraceKey\\\": \\\"stacktrace\\\",\\n    \\\"lineEnding\\\": \\\"\\\",\\n    \\\"levelEncoder\\\": \\\"\\\",\\n    \\\"timeEncoder\\\": \\\"iso8601\\\",\\n    \\\"durationEncoder\\\": \\\"\\\",\\n    \\\"callerEncoder\\\": \\\"\\\"\\n  }\\n}\"}"
+                            },
+                            {
+                                "name": "K_METRICS_CONFIG",
+                                "value": "{\"Domain\":\"triggermesh.io/sources\",\"Component\":\"awssnssource\",\"PrometheusPort\":0,\"ConfigMap\":{}}"
+                            }
+                        ],
+                        "image": "gcr.io/triggermesh/awssnssource",
+                        "name": "adapter",
+                        "readinessProbe": {
+                            "successThreshold": 1,
+                            "tcpSocket": {
+                                "port": 0
+                            }
+                        },
+                        "resources": {}
+                    }
+                ],
+                "timeoutSeconds": 300
+            }
+        },
+        "traffic": [
+            {
+                "latestRevision": true,
+                "percent": 100
+            }
+        ]
+    },
+    "status": {
+        "conditions": [
+            {
+                "lastTransitionTime": "2020-04-23T21:14:02Z",
+                "message": "Revision \"awssnssource-sample-6wfhj\" failed with message: Unable to fetch image \"gcr.io/triggermesh/awssnssource\": failed to resolve image to digest: failed to fetch image information: GET https://gcr.io/v2/triggermesh/awssnssource/manifests/latest: MANIFEST_UNKNOWN: Failed to fetch \"latest\" from request \"/v2/triggermesh/awssnssource/manifests/latest\"..",
+                "reason": "RevisionFailed",
+                "status": "False",
+                "type": "ConfigurationsReady"
+            },
+            {
+                "lastTransitionTime": "2020-04-23T21:14:02Z",
+                "message": "Configuration \"awssnssource-sample\" does not have any ready Revision.",
+                "reason": "RevisionMissing",
+                "status": "False",
+                "type": "Ready"
+            },
+            {
+                "lastTransitionTime": "2020-04-23T21:14:02Z",
+                "message": "Configuration \"awssnssource-sample\" does not have any ready Revision.",
+                "reason": "RevisionMissing",
+                "status": "False",
+                "type": "RoutesReady"
+            }
+        ],
+        "latestCreatedRevisionName": "awssnssource-sample-6wfhj",
+        "observedGeneration": 1,
+        "url": "http://awssnssource-sample.dev.example.com"
+    }
+}