[WIP] TCP Convergence Layer Protocol Version 4

cla/convergence_layer.go

@@ -1,4 +1,4 @@
-// Package cla defines two interfaces for convergence layers.
+// Package cla defines two interfaces for Convergence Layer Adapters.
 //
 // The ConvergenceReceiver specifies a type which receives bundles and forwards
 // those to an exposed channel.
@@ -9,23 +9,38 @@
 // An implemented convergence layer can be a ConvergenceReceiver,
 // ConvergenceSender or even both. This depends on the convergence layer's
 // specification and is an implemention matter.
+//
+// Furthermore, the ConvergenceProvider provides the ability to create new
+// instances of Convergence objects.
+//
+// Those types are generalized by the Convergable interface.
+//
+// A centralized instance for CLA management offers the Manager, designed to
+// work seamlessly with the types above.
 package cla
 
 import "github.com/dtn7/dtn7-go/bundle"
 
+// Convergable describes any kind of type which supports convergence layer-
+// related services. This can be both a more specified Convergence interface
+// type or a ConvergenceProvider.
+type Convergable interface {
+	// Close signals this Convergable to shut down.
+	Close()
+}
+
 // Convergence is an interface to describe all kinds of Convergence Layer
 // Adapters. There should not be a direct implemention of this interface. One
 // must implement ConvergenceReceiver and/or ConvergenceSender, which are both
 // extending this interface.
 // A type can be both a ConvergenceReceiver and ConvergenceSender.
 type Convergence interface {
+	Convergable
+
 	// Start starts this Convergence{Receiver,Sender} and might return an error
 	// and a boolean indicating if another Start should be tried later.
 	Start() (error, bool)
 
-	// Close signals this Convergence{Receiver,Send} to shut down.
-	Close()
-
 	// Channel represents a return channel for transmitted bundles, status
 	// messages, etc.
 	Channel() chan ConvergenceStatus
@@ -62,3 +77,20 @@ type ConvergenceSender interface {
 	// if it's known. Otherwise the zero endpoint will be returned.
 	GetPeerEndpointID() bundle.EndpointID
 }
+
+// ConvergenceProvider is a more general kind of CLA service which does not
+// transfer any Bundles by itself, but supplies/creates new Convergence types.
+// Those Convergence objects will be passed to a Manager. Thus, one might think
+// of a ConvergenceProvider as some kind of middleware.
+type ConvergenceProvider interface {
+	Convergable
+
+	// RegisterManager tells the ConvergenceProvider where to report new instances
+	// of Convergence to.
+	RegisterManager(*Manager)
+
+	// Start starts this ConvergenceProvider. Before being started, the the
+	// RegisterManager method tells this ConvergenceProvider its Manager. However,
+	// the Manager will both call the RegisterManager and Start methods.
+	Start() error
+}

cla/manager.go

@@ -24,6 +24,11 @@ type Manager struct {
 	// convs: Map[string]*convergenceElem
 	convs *sync.Map
 
+	// providers is an array of ConvergenceProvider. Those will report their
+	// created Convergence objects to this Manager, which also supervises it.
+	providers      []ConvergenceProvider
+	providersMutex sync.Mutex
+
 	// inChnl receives ConvergenceStatus while outChnl passes it on. Both channels
 	// are not buffered. While this is not a problem for inChnl, outChnl must
 	// always be read, otherwise the Manager will block.
@@ -48,7 +53,7 @@ func NewManager() *Manager {
 
 		convs: new(sync.Map),
 
-		inChnl:  make(chan ConvergenceStatus),
+		inChnl:  make(chan ConvergenceStatus, 100),
 		outChnl: make(chan ConvergenceStatus),
 
 		stopSyn: make(chan struct{}),
@@ -70,13 +75,19 @@ func (manager *Manager) handler() {
 	for {
 		select {
 		case <-manager.stopSyn:
-			log.Debug("CLA Manager received closing-signal")
+			log.Debug("CLA Manager received closing signal")
 
 			manager.convs.Range(func(_, convElem interface{}) bool {
 				manager.Unregister(convElem.(*convergenceElem).conv)
 				return true
 			})
 
+			manager.providersMutex.Lock()
+			for _, provider := range manager.providers {
+				provider.Close()
+			}
+			manager.providersMutex.Unlock()
+
 			close(manager.inChnl)
 			close(manager.outChnl)
 
@@ -146,12 +157,22 @@ func (manager *Manager) Close() {
 	<-manager.stopAck
 }
 
-// Register a new CLA.
-func (manager *Manager) Register(conv Convergence) {
+// Register any kind of Convergable.
+func (manager *Manager) Register(conv Convergable) {
 	if manager.isStopped() {
 		return
 	}
 
+	if c, ok := conv.(Convergence); ok {
+		manager.registerConvergence(c)
+	} else if c, ok := conv.(ConvergenceProvider); ok {
+		manager.registerProvider(c)
+	} else {
+		log.WithField("convergence", conv).Warn("Unknown kind of Convergable")
+	}
+}
+
+func (manager *Manager) registerConvergence(conv Convergence) {
 	// Check if this CLA is already known. Re-activate a deactivated CLA or abort.
 	var ce *convergenceElem
 	if convElem, exists := manager.convs.Load(conv.Address()); exists {
@@ -192,8 +213,38 @@ func (manager *Manager) Register(conv Convergence) {
 	}
 }
 
-// Unregister an already known CLA.
-func (manager *Manager) Unregister(conv Convergence) {
+func (manager *Manager) registerProvider(conv ConvergenceProvider) {
+	manager.providersMutex.Lock()
+	defer manager.providersMutex.Unlock()
+
+	for _, provider := range manager.providers {
+		if conv == provider {
+			log.WithField("provider", conv).Debug("Provider registration aborted, already known")
+			return
+		}
+	}
+
+	manager.providers = append(manager.providers, conv)
+
+	conv.RegisterManager(manager)
+
+	if err := conv.Start(); err != nil {
+		log.WithError(err).WithField("provider", conv).Warn("Starting Provider errored")
+	}
+}
+
+// Unregister any kind of Convergable.
+func (manager *Manager) Unregister(conv Convergable) {
+	if c, ok := conv.(Convergence); ok {
+		manager.unregisterConvergence(c)
+	} else if c, ok := conv.(ConvergenceProvider); ok {
+		manager.unregisterProvider(c)
+	} else {
+		log.WithField("convergence", conv).Warn("Unknown kind of Convergable")
+	}
+}
+
+func (manager *Manager) unregisterConvergence(conv Convergence) {
 	convElem, exists := manager.convs.Load(conv.Address())
 	if !exists {
 		log.WithFields(log.Fields{
@@ -208,8 +259,20 @@ func (manager *Manager) Unregister(conv Convergence) {
 	manager.convs.Delete(conv.Address())
 }
 
-// Restart an already known CLA.
-func (manager *Manager) Restart(conv Convergence) {
+func (manager *Manager) unregisterProvider(conv ConvergenceProvider) {
+	manager.providersMutex.Lock()
+	defer manager.providersMutex.Unlock()
+
+	for i, provider := range manager.providers {
+		if conv == provider {
+			manager.providers = append(manager.providers[:i], manager.providers[i+1:]...)
+			return
+		}
+	}
+}
+
+// Restart a known Convergable.
+func (manager *Manager) Restart(conv Convergable) {
 	manager.Unregister(conv)
 	manager.Register(conv)
 }

cla/tcpcl/0doc.go

@@ -0,0 +1,6 @@
+// Package tcpcl provides a library for the Delay-Tolerant Networking TCP Convergence Layer Protocol Version 4,
+// draft-ietf-dtn-tcpclv4-14.
+//
+// A new TCPCL server can be started by the Listener, which provides multiple connection to its Clients. To reach
+// a remote server, a new Client connection can be dialed.
+package tcpcl