added queue and exchange existance verification to source controller to avoid not existing errors (#961)

Co-authored-by: gabriel <[email protected]>

Author: knative-prow-robot

pkg/rabbit/types.go

@@ -21,6 +21,8 @@ import (
 	"net/url"
 
 	rabbitv1beta1 "knative.dev/eventing-rabbitmq/third_party/pkg/apis/rabbitmq.com/v1beta1"
+
+	rmqv1beta1 "knative.dev/eventing-rabbitmq/third_party/pkg/client/clientset/versioned/typed/rabbitmq.com/v1beta1"
 )
 
 type Result struct {
@@ -29,6 +31,7 @@ type Result struct {
 }
 
 type Service interface {
+	RabbitmqV1beta1() rmqv1beta1.RabbitmqV1beta1Interface
 	RabbitMQURL(context.Context, *rabbitv1beta1.RabbitmqClusterReference) (*url.URL, error)
 	ReconcileExchange(context.Context, *ExchangeArgs) (Result, error)
 	ReconcileQueue(context.Context, *QueueArgs) (Result, error)

pkg/reconciler/source/rabbitmqsource.go

@@ -154,6 +154,18 @@ func (r *Reconciler) reconcileRabbitObjects(ctx context.Context, src *v1alpha1.R
 	src.Status.MarkSecretReady()
 
 	if src.Spec.RabbitmqResourcesConfig.Predeclared {
+		_, err := r.rabbit.RabbitmqV1beta1().Exchanges(src.Namespace).Get(ctx, src.Spec.RabbitmqResourcesConfig.ExchangeName, metav1.GetOptions{})
+		if apierrors.IsNotFound(err) {
+			logger.Error("exchange not found", "exchange", src.Spec.RabbitmqResourcesConfig.ExchangeName)
+			return err
+		}
+
+		_, err = r.rabbit.RabbitmqV1beta1().Queues(src.Namespace).Get(ctx, src.Spec.RabbitmqResourcesConfig.QueueName, metav1.GetOptions{})
+		if apierrors.IsNotFound(err) {
+			logger.Error("queue not found", "queue", src.Spec.RabbitmqResourcesConfig.QueueName)
+			return err
+		}
+
 		logger.Info("predeclared set to true; no RabbitMQ objects to reconcile",
 			"source", src.Name,
 			"predeclared queue", src.Spec.RabbitmqResourcesConfig.QueueName)