Merge pull request #130 from christopherhein/chore/21-single-queue

Single Queue With Multiple Topics Subscribed

Author: Christopher Hein

cloudformation/dynamodb.yaml

@@ -1,5 +1,5 @@
 AWSTemplateFormatVersion: 2010-09-09
-Description: 'AWS Service Operator - Amazon DynamoDB'
+Description: 'AWS Service Operator - Amazon DynamoDB (aso-1otq2cgc4)'
 Parameters:
   Namespace:
     Description: >-

cloudformation/ecrrepository.yaml

@@ -1,5 +1,5 @@
 AWSTemplateFormatVersion: 2010-09-09
-Description: 'AWS Service Operator - Amazon ECR Repository'
+Description: 'AWS Service Operator - Amazon ECR Repository (aso-1otq2cgc9)'
 Parameters:
   Namespace:
     Description: >-

cloudformation/s3bucket.yaml

@@ -1,5 +1,5 @@
 AWSTemplateFormatVersion: 2010-09-09
-Description: 'AWS Service Operator - Amazon S3 Bucket'
+Description: 'AWS Service Operator - Amazon S3 Bucket (aso-1otq2cgce)'
 Parameters:
   Namespace:
     Description: >-

cloudformation/snssubscription.yaml

@@ -1,5 +1,5 @@
 AWSTemplateFormatVersion: 2010-09-09
-Description: 'AWS Service Operator - Amazon SNS Subscription'
+Description: 'AWS Service Operator - Amazon SNS Subscription (aso-1otq2cgcj)'
 Parameters:
   Namespace:
     Description: >-

cloudformation/snstopic.yaml

@@ -1,5 +1,5 @@
 AWSTemplateFormatVersion: 2010-09-09
-Description: 'AWS Service Operator - Amazon SNS Topic'
+Description: 'AWS Service Operator - Amazon SNS Topic (aso-1otq2cgcp)'
 Parameters:
   Namespace:
     Description: >-

cloudformation/sqsqueue.yaml

@@ -1,5 +1,5 @@
 AWSTemplateFormatVersion: 2010-09-09
-Description: "AWS Service Operator - Amazon SQS Queue"
+Description: "AWS Service Operator - Amazon SQS Queue (aso-1otq2cgcu)"
 Parameters:
   Namespace:
     Type: String

code-generation/pkg/codegen/assets/base.go.templ

@@ -3,28 +3,37 @@ package base
 import (
   "context"
   "github.com/awslabs/aws-service-operator/pkg/config"
+  "github.com/awslabs/aws-service-operator/pkg/queuemanager"
   {{- range $index, $element := .Items}}
   "github.com/awslabs/aws-service-operator/pkg/operators/{{$element.Spec.Resource.Name}}"
   {{- end}}
 )
 
 type base struct {
   config *config.Config
+  queueManager *queuemanager.QueueManager
+  {{- range $index, $element := .Items}}
+  {{$element.Spec.Resource.Name}} *{{$element.Spec.Resource.Name}}.Operator
+  {{- end}}
 }
 
 func New(
   config *config.Config,
+  queueManager *queuemanager.QueueManager,
 ) *base {
   return &base{
     config: config,
+    queueManager: queueManager,
+    {{- range $index, $element := .Items}}
+    {{$element.Spec.Resource.Name}}: {{$element.Spec.Resource.Name}}.NewOperator(config, queueManager),
+    {{- end}}
   }
 }
 
 func (b *base) Watch(ctx context.Context, namespace string) {
   {{- range $index, $element := .Items}}
   if b.config.Resources["{{$element.Spec.Resource.Name}}"] {
-    {{$element.Spec.Resource.Name}}operator := {{$element.Spec.Resource.Name}}.NewOperator(b.config)
-  	go {{$element.Spec.Resource.Name}}operator.StartWatch(ctx, namespace)
+  	go b.{{$element.Spec.Resource.Name}}.StartWatch(ctx, namespace)
   }
 	{{- end}}
 

code-generation/pkg/codegen/assets/operator.go.templ

@@ -14,12 +14,13 @@ import (
 	{{- end}}
 
 	"github.com/awslabs/aws-service-operator/pkg/config"
-	{{- if .Spec.Queue}}
-    "github.com/awslabs/aws-service-operator/pkg/queue"
-    corev1 "k8s.io/api/core/v1"
-    "github.com/iancoleman/strcase"
-    "strings"
-		awsclient "github.com/awslabs/aws-service-operator/pkg/client/clientset/versioned/typed/service-operator.aws/v1alpha1"
+  "github.com/awslabs/aws-service-operator/pkg/queuemanager"
+  {{- if .Spec.Queue}}
+  corev1 "k8s.io/api/core/v1"
+  "github.com/iancoleman/strcase"
+  "strings"
+	awsclient "github.com/awslabs/aws-service-operator/pkg/client/clientset/versioned/typed/service-operator.aws/v1alpha1"
+	"github.com/awslabs/aws-service-operator/pkg/queue"
 	{{- end}}
 	"github.com/awslabs/aws-service-operator/pkg/operator"
 	"k8s.io/client-go/tools/cache"
@@ -34,12 +35,23 @@ import (
 type Operator struct {
 	config       *config.Config
 	topicARN     string
+	queueManager *queuemanager.QueueManager
 }
 
 // NewOperator create controller for watching object store custom resources created
-func NewOperator(config *config.Config) *Operator {
+func NewOperator(config *config.Config, queueManager *queuemanager.QueueManager) *Operator {
+	{{- if .Spec.Queue}}
+	queuectrl := queue.New(config, config.AWSClientset, 10)
+	topicARN, _ := queuectrl.Register("{{.Spec.Resource.Name}}")
+	queueManager.Add(topicARN, queuemanager.HandlerFunc(QueueUpdater))
+	{{- end}}
+
 	return &Operator{
 		config:       config,
+		{{- if .Spec.Queue}}
+		topicARN: topicARN,
+		{{- end}}
+		queueManager: queueManager,
 	}
 }
 
@@ -50,19 +62,14 @@ func (c *Operator) StartWatch(ctx context.Context, namespace string) {
 		UpdateFunc: c.onUpdate,
 		DeleteFunc: c.onDelete,
 	}
-	{{- if .Spec.Queue}}
-	queuectrl := queue.New(c.config, c.config.AWSClientset, 1)
-	c.topicARN, _, _, _ = queuectrl.Register("{{.Spec.Resource.Name}}", &awsV1alpha1.{{.Spec.Kind}}{})
-	go queuectrl.StartWatch(queue.HandlerFunc(QueueUpdater), ctx.Done())
-	{{- end}}
 
 	oper := operator.New("{{.Spec.Resource.Plural}}", namespace, resourceHandlers, c.config.AWSClientset.RESTClient())
 	oper.Watch(&awsV1alpha1.{{.Spec.Kind}}{}, ctx.Done())
 }
 
 {{- if .Spec.Queue}}
 // QueueUpdater will take the messages from the queue and process them
-func QueueUpdater(config *config.Config, msg *queue.MessageBody) error {
+func QueueUpdater(config *config.Config, msg *queuemanager.MessageBody) error {
 	logger := config.Logger
 	var name, namespace string
 	if msg.Updatable {

code-generation/pkg/codegen/templates.go

@@ -150,13 +150,15 @@ items:
     - pods
     - configmaps
     - services
+    - events
     verbs:
     - get
     - list
     - watch
     - create
     - delete
     - update
+    - patch
   - apiGroups:
     - apiextensions.k8s.io
     resources:
@@ -203,7 +205,7 @@ items:
     template:
       metadata:
         annotations:
-          iam.amazonaws.com/role: arn:aws:iam::<ACCOUNT_ID>:role/aws-service-operator
+          iam.amazonaws.com/role: aws-service-operator
         labels:
           app: aws-service-operator
       spec: