Bus monitor polish

config/buses/stub/stub-bus.yaml

@@ -18,7 +18,7 @@ metadata:
 spec:
   dispatcher:
     name: dispatcher
-    image: github.com/knative/eventing/pkg/buses/stub
+    image: github.com/knative/eventing/pkg/buses/stub/dispatcher
     args: [
       "-logtostderr",
       "-stderrthreshold", "INFO",

pkg/apis/channels/v1alpha1/subscription_validation.go

@@ -27,12 +27,16 @@ func (s *Subscription) Validate() *apis.FieldError {
 }
 
 func (ss *SubscriptionSpec) Validate() *apis.FieldError {
-	if len(ss.Channel) == 0 {
+	if ss.Channel == "" {
 		fe := apis.ErrMissingField("channel")
 		fe.Details = "the Subscription must reference a Channel"
 		return fe
 	}
-	// TODO: should subscriptions have a subscriber?
+	if ss.Subscriber == "" {
+		fe := apis.ErrMissingField("subscriber")
+		fe.Details = "the Subscription must reference a Subscriber"
+		return fe
+	}
 	return nil
 }
 

pkg/apis/channels/v1alpha1/subscription_validation_test.go

@@ -16,9 +16,10 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"testing"
+
 	"github.com/google/go-cmp/cmp"
 	"github.com/knative/pkg/apis"
-	"testing"
 )
 
 func TestSubscriptionSpecValidation(t *testing.T) {
@@ -28,25 +29,29 @@ func TestSubscriptionSpecValidation(t *testing.T) {
 		want *apis.FieldError
 	}{{
 		name: "valid",
-		c: &SubscriptionSpec{
-			Channel: "foo",
-		},
-		want: nil,
-	}, {
-		name: "valid with subscriber",
 		c: &SubscriptionSpec{
 			Channel:    "bar",
 			Subscriber: "foo",
 		},
 		want: nil,
 	}, {
-		name: "valid with subscriber and arguments",
+		name: "valid with arguments",
 		c: &SubscriptionSpec{
 			Channel:    "bar",
 			Subscriber: "foo",
 			Arguments:  &[]Argument{{Name: "foo", Value: "bar"}},
 		},
 		want: nil,
+	}, {
+		name: "missing subscriber",
+		c: &SubscriptionSpec{
+			Channel: "foo",
+		},
+		want: func() *apis.FieldError {
+			fe := apis.ErrMissingField("subscriber")
+			fe.Details = "the Subscription must reference a Subscriber"
+			return fe
+		}(),
 	}, {
 		name: "empty",
 		c:    &SubscriptionSpec{},

pkg/buses/bus.go

@@ -0,0 +1,169 @@
+/*
+ * Copyright 2018 The Knative Authors
+ *
+ * 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 buses
+
+import (
+	"fmt"
+
+	channelsv1alpha1 "github.com/knative/eventing/pkg/apis/channels/v1alpha1"
+)
+
+type bus struct {
+	busRef       BusReference
+	handlerFuncs EventHandlerFuncs
+
+	reconciler *Reconciler
+	cache      *Cache
+	dispatcher *MessageDispatcher
+	receiver   *MessageReceiver
+}
+
+// BusOpts holds configuration options for new buses. These options are not
+// required for proper operation of the bus, but are useful to override the
+// default behavior and for testing.
+type BusOpts struct {
+	// MasterURL is the address of the Kubernetes API server. Overrides any
+	// value in kubeconfig. Only required if out-of-cluster.
+	MasterURL string
+	// KubeConfig is the path to a kubeconfig. Only required if out-of-cluster.
+	KubeConfig string
+
+	// Cache to use for this bus. A cache will be created for the bus if not
+	// specified.
+	Cache *Cache
+	// Reconciler to use for this bus. A reconciler wil be created for the bus
+	// if not specified.
+	Reconciler *Reconciler
+	// MessageDispatcher to use for this bus. The message dispatcher is used to
+	// send messages from the bus to a subscriber. A message dispatcher will be
+	// created for the bus if needed and not specified.
+	MessageDispatcher *MessageDispatcher
+	// MessageReceiver to use for this bus. The message receiver is used to
+	// receive message sent to a channel. A message receiver will be created
+	// for the bus if needed and not specified.
+	MessageReceiver *MessageReceiver
+}
+
+// BusProvisioner provisions channels and subscriptions for a bus on backing
+// infrastructure.
+type BusProvisioner interface {
+	Run(threadiness int, stopCh <-chan struct{})
+}
+
+// NewBusProvisioner creates a new provisioner for a specific bus.
+// EventHandlerFuncs are used to be notified when a channel or subscription is
+// created, updated or removed.
+func NewBusProvisioner(busRef BusReference, handlerFuncs EventHandlerFuncs, opts *BusOpts) BusProvisioner {
+	if opts == nil {
+		opts = &BusOpts{}
+	}
+	if opts.Cache == nil {
+		opts.Cache = NewCache()
+	}
+	if opts.Reconciler == nil {
+		opts.Reconciler = NewReconciler(Provisioner, opts.MasterURL, opts.KubeConfig, opts.Cache, handlerFuncs)
+	}
+
+	return &bus{
+		busRef:       busRef,
+		handlerFuncs: handlerFuncs,
+		cache:        opts.Cache,
+		reconciler:   opts.Reconciler,
+	}
+}
+
+// BusDispatcher dispatches messages from channels to subscribers via backing
+// infrastructure.
+type BusDispatcher interface {
+	Run(threadiness int, stopCh <-chan struct{})
+	DispatchMessage(subscriptionRef SubscriptionReference, message *Message) error
+}
+
+// NewBusDispatcher creates a new dispatcher for a specific bus.
+// EventHandlerFuncs are used to be notified when a subscription is created,
+// updated or removed, or a message is received.
+func NewBusDispatcher(busRef BusReference, handlerFuncs EventHandlerFuncs, opts *BusOpts) BusDispatcher {
+	var b *bus
+
+	if opts == nil {
+		opts = &BusOpts{}
+	}
+	if opts.Cache == nil {
+		opts.Cache = NewCache()
+	}
+	if opts.Reconciler == nil {
+		opts.Reconciler = NewReconciler(Dispatcher, opts.MasterURL, opts.KubeConfig, opts.Cache, handlerFuncs)
+	}
+	if opts.MessageDispatcher == nil {
+		opts.MessageDispatcher = NewMessageDispatcher()
+	}
+	if opts.MessageReceiver == nil {
+		opts.MessageReceiver = NewMessageReceiver(func(channelRef ChannelReference, message *Message) error {
+			return b.receiveMessage(channelRef, message)
+		})
+	}
+
+	b = &bus{
+		busRef:       busRef,
+		handlerFuncs: handlerFuncs,
+
+		cache:      opts.Cache,
+		reconciler: opts.Reconciler,
+		dispatcher: opts.MessageDispatcher,
+		receiver:   opts.MessageReceiver,
+	}
+
+	return b
+}
+
+// Run starts the bus's processing.
+func (b bus) Run(threadiness int, stopCh <-chan struct{}) {
+	go b.reconciler.Run(b.busRef, threadiness, stopCh)
+	b.reconciler.WaitForCacheSync(stopCh)
+	if b.receiver != nil {
+		go b.receiver.Run(stopCh)
+	}
+
+	<-stopCh
+}
+
+func (b *bus) receiveMessage(channelRef ChannelReference, message *Message) error {
+	_, err := b.cache.Channel(channelRef)
+	if err != nil {
+		return ErrUnknownChannel
+	}
+	return b.handlerFuncs.onReceiveMessage(channelRef, message)
+}
+
+// DispatchMessage sends a message to a subscriber. This function is only
+// avilable for bus dispatchers.
+func (b *bus) DispatchMessage(subscriptionRef SubscriptionReference, message *Message) error {
+	subscription, err := b.cache.Subscription(subscriptionRef)
+	if err != nil {
+		return fmt.Errorf("unable to dispatch to unknown subscription %q", subscriptionRef.String())
+	}
+	return b.dispatchMessage(subscription, message)
+}
+
+func (b *bus) dispatchMessage(subscription *channelsv1alpha1.Subscription, message *Message) error {
+	subscriber := subscription.Spec.Subscriber
+	defaults := DispatchDefaults{
+		Namespace: subscription.Namespace,
+		ReplyTo:   subscription.Spec.ReplyTo,
+	}
+	return b.dispatcher.DispatchMessage(message, subscriber, defaults)
+}

pkg/buses/cache.go

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2018 The Knative Authors
+ *
+ * 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 buses
+
+import (
+	"fmt"
+
+	channelsv1alpha1 "github.com/knative/eventing/pkg/apis/channels/v1alpha1"
+)
+
+// NewCache create a cache that is able to save and retrive Channels and
+// Subscriptions by their reference.
+func NewCache() *Cache {
+	return &Cache{
+		channels:      make(map[ChannelReference]*channelsv1alpha1.Channel),
+		subscriptions: make(map[SubscriptionReference]*channelsv1alpha1.Subscription),
+	}
+}
+
+// Cache able to save and retrive Channels and Subscriptions by their
+// reference. It is used by the reconciler to track which resources have been
+// provisioned and comparing updated resources to the provisioned version.
+type Cache struct {
+	channels      map[ChannelReference]*channelsv1alpha1.Channel
+	subscriptions map[SubscriptionReference]*channelsv1alpha1.Subscription
+}
+
+// Channel returns a cached channel for provided reference or an error if the
+// channel is not in the cache.
+func (c *Cache) Channel(channelRef ChannelReference) (*channelsv1alpha1.Channel, error) {
+	channel, ok := c.channels[channelRef]
+	if !ok {
+		return nil, fmt.Errorf("unknown channel %q", channelRef.String())
+	}
+	return channel, nil
+}
+
+// Subscription returns a cached subscription for provided reference or an
+// error if the subscription is not in the cache.
+func (c *Cache) Subscription(subscriptionRef SubscriptionReference) (*channelsv1alpha1.Subscription, error) {
+	subscription, ok := c.subscriptions[subscriptionRef]
+	if !ok {
+		return nil, fmt.Errorf("unknown subscription %q", subscriptionRef.String())
+	}
+	return subscription, nil
+}
+
+// AddChannel adds, or updates, the provided channel to the cache for later
+// retrieal by its reference.
+func (c *Cache) AddChannel(channel *channelsv1alpha1.Channel) {
+	if channel == nil {
+		return
+	}
+	channelRef := NewChannelReference(channel)
+	c.channels[channelRef] = channel
+}
+
+// RemoveChannel removes the provided channel from the cache.
+func (c *Cache) RemoveChannel(channel *channelsv1alpha1.Channel) {
+	if channel == nil {
+		return
+	}
+	channelRef := NewChannelReference(channel)
+	delete(c.channels, channelRef)
+}
+
+// AddSubscription adds, or updates, the provided subscription to the cache for
+// later retrieal by its reference.
+func (c *Cache) AddSubscription(subscription *channelsv1alpha1.Subscription) {
+	if subscription == nil {
+		return
+	}
+	subscriptionRef := NewSubscriptionReference(subscription)
+	c.subscriptions[subscriptionRef] = subscription
+}
+
+// RemoveSubscription removes the provided subscription from the cache.
+func (c *Cache) RemoveSubscription(subscription *channelsv1alpha1.Subscription) {
+	if subscription == nil {
+		return
+	}
+	subscriptionRef := NewSubscriptionReference(subscription)
+	delete(c.subscriptions, subscriptionRef)
+}