[core] Add support for manually configured outgoing channels

Author: teo

core/task/channel/inbound.go

@@ -50,7 +50,11 @@ chans.data1.0.type          = push
 chans.data1.numSockets      = 1
  */
 
-func (inbound Inbound) ToFMQMap(port uint64) (pm controlcommands.PropertyMap) {
+func (inbound *Inbound) ToFMQMap(port uint64) (pm controlcommands.PropertyMap) {
+	return inbound.buildFMQMap(fmt.Sprintf("tcp://*:%d", port))
+}
+
+func (inbound *Inbound) buildFMQMap(address string) (pm controlcommands.PropertyMap) {
 	pm = make(controlcommands.PropertyMap)
 	const chans = "chans"
 	chName := inbound.Name
@@ -59,7 +63,7 @@ func (inbound Inbound) ToFMQMap(port uint64) (pm controlcommands.PropertyMap) {
 	prefix := strings.Join([]string{chans, chName, "0"}, ".")
 
 	chanProps := controlcommands.PropertyMap{
-		"address": fmt.Sprintf("tcp://*:%d", port),
+		"address": address,
 		"method": "bind",
 		"rateLogging": strconv.Itoa(inbound.RateLogging),
 		"rcvBufSize": strconv.Itoa(inbound.RcvBufSize),

core/task/channel/outbound.go

@@ -79,7 +79,37 @@ chans.data1.0.type          = pull
 chans.data1.numSockets      = 1
 */
 
-func (outbound *Outbound) ToFMQMap(endpoint Endpoint) (pm controlcommands.PropertyMap) {
+func (outbound *Outbound) ToFMQMap(bindMap BindMap) (pm controlcommands.PropertyMap) {
+	if outbound == nil {
+		return
+	}
+
+	var address string
+	// If an explicit target was provided, we use it
+	if strings.HasPrefix(outbound.Target, "tcp://") ||
+		strings.HasPrefix(outbound.Target, "ipc://") {
+		address = outbound.Target
+	} else {
+		// we don't need class.Bind data for this one, only task.bindPorts after resolving paths!
+		for chPath, endpoint := range bindMap {
+			// FIXME: implement more sophisticated channel matching here
+			if outbound.Target == chPath {
+
+				// We have a match, so we generate a resolved target address and break
+				address = fmt.Sprintf("tcp://%s:%d", endpoint.Host, endpoint.Port)
+				break
+			}
+		}
+	}
+
+	if len(address) == 0 {
+		return
+	}
+
+	return outbound.buildFMQMap(address)
+}
+
+func (outbound *Outbound) buildFMQMap(address string) (pm controlcommands.PropertyMap) {
 	pm = make(controlcommands.PropertyMap)
 	const chans = "chans"
 	chName := outbound.Name
@@ -88,7 +118,7 @@ func (outbound *Outbound) ToFMQMap(endpoint Endpoint) (pm controlcommands.Proper
 	prefix := strings.Join([]string{chans, chName, "0"}, ".")
 
 	chanProps := controlcommands.PropertyMap{
-		"address": fmt.Sprintf("tcp://%s:%d", endpoint.Host, endpoint.Port),
+		"address": address,
 		"method": "connect",
 		"rateLogging": strconv.Itoa(outbound.RateLogging),
 		"rcvBufSize": strconv.Itoa(outbound.RcvBufSize),

core/task/task.go

@@ -222,19 +222,14 @@ func (t Task) BuildPropertyMap(bindMap channel.BindMap) controlcommands.Property
 				}
 			}
 
-			for _, outbCh := range t.parent.CollectOutboundChannels() {
-				// we don't need class.Bind data for this one, only task.bindPorts after resolving paths!
-				for chPath, endpoint := range bindMap {
-					// FIXME: implement more sophisticated channel matching here
-					if outbCh.Target == chPath {
-
-						// We get the FairMQ-formatted propertyMap from the outbound channel spec
-						chanProps := outbCh.ToFMQMap(endpoint)
-
-						// And we copy it into the task's propertyMap
-						for k, v := range chanProps {
-							propMap[k] = v
-						}
+			for _, outboundCh := range t.parent.CollectOutboundChannels() {
+				// We get the FairMQ-formatted propertyMap from the outbound channel spec
+				chanProps := outboundCh.ToFMQMap(bindMap)
+
+				// And if valid, we copy it into the task's propertyMap
+				if len(chanProps) > 0 {
+					for k, v := range chanProps {
+						propMap[k] = v
 					}
 				}
 			}