Multiple connectors for a single extractor #120

Author: atomashpolskiy

RELEASE-NOTES.txt

@@ -4,6 +4,7 @@ Release 2.2
 #114 Targets mapped event before merging update
 #116 tokenManager is null in DefaultDeleteBuilder
 #117 JdbcNormalizer for java.time classes
+#120 Multiple connectors for a single extractor
 
 Release 2.1
 

link-move-csv/src/test/java/com/nhl/link/move/runtime/extractor/BaseExtractorFactoryTest.java

@@ -34,7 +34,7 @@ public void setUpExtractorFactory() {
 	@Before
 	public void setUpExtractorModel() {
 		model = new MutableExtractorModel("testExtractorConfig");
-		model.setConnectorId(CONNECTOR_ID);
+		model.addConnectorId(CONNECTOR_ID);
 		model.setAttributes(new BaseRowAttribute[0]);
 	}
 

link-move/src/main/java/com/nhl/link/move/extractor/model/ContainerAwareExtractorModel.java

@@ -1,5 +1,6 @@
 package com.nhl.link.move.extractor.model;
 
+import java.util.Collection;
 import java.util.Map;
 
 import com.nhl.link.move.RowAttribute;
@@ -43,6 +44,11 @@ public String getConnectorId() {
 		return connectorId != null ? connectorId : parent.getConnectorId();
 	}
 
+	@Override
+	public Collection<String> getConnectorIds() {
+		return delegate.getConnectorIds().isEmpty() ? parent.getConnectorIds() : delegate.getConnectorIds();
+	}
+
 	@Override
 	public Map<String, String> getProperties() {
 		return delegate.getProperties();

link-move/src/main/java/com/nhl/link/move/extractor/model/ExtractorModel.java

@@ -1,5 +1,6 @@
 package com.nhl.link.move.extractor.model;
 
+import java.util.Collection;
 import java.util.Map;
 
 import com.nhl.link.move.RowAttribute;
@@ -12,14 +13,21 @@
  */
 public interface ExtractorModel {
 
-	public static final String DEFAULT_NAME = "default_extractor";
+	String DEFAULT_NAME = "default_extractor";
 
 	String getName();
 
 	String getType();
 
+	@Deprecated
 	String getConnectorId();
 
+	/**
+	 * @return Collection of connector IDs
+	 * @since 2.2
+     */
+	Collection<String> getConnectorIds();
+
 	Map<String, String> getProperties();
 
 	RowAttribute[] getAttributes();

link-move/src/main/java/com/nhl/link/move/extractor/model/ExtractorModelContainer.java

@@ -15,8 +15,15 @@ public interface ExtractorModelContainer {
 
 	String getType();
 
+	@Deprecated
 	String getConnectorId();
 
+	/**
+	 * @return Collection of connector IDs
+	 * @since 2.2
+     */
+	Collection<String> getConnectorIds();
+
 	ExtractorModel getExtractor(String name);
 
 	Collection<String> getExtractorNames();

link-move/src/main/java/com/nhl/link/move/extractor/model/MutableExtractorModel.java

@@ -1,7 +1,11 @@
 package com.nhl.link.move.extractor.model;
 
+import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 
 import com.nhl.link.move.RowAttribute;
 
@@ -12,13 +16,14 @@ public class MutableExtractorModel implements ExtractorModel {
 
 	private String name;
 	private String type;
-	private String connectorId;
+	private Set<String> connectorIds;
 	private long loadedOn;
 	private RowAttribute[] attributes;
 	private Map<String, String> properties;
 
 	public MutableExtractorModel(String name) {
 		this.name = name;
+		this.connectorIds = new HashSet<>();
 		this.properties = new HashMap<>();
 	}
 
@@ -39,7 +44,18 @@ public String getType() {
 
 	@Override
 	public String getConnectorId() {
-		return connectorId;
+		if (connectorIds.isEmpty()) {
+			// connector IDs can be missing if this model is a part of a model container
+			return null;
+		} if (connectorIds.size() == 1) {
+			return connectorIds.iterator().next();
+		}
+		throw new IllegalStateException("Multiple connector IDs specified in model");
+	}
+
+	@Override
+	public Collection<String> getConnectorIds() {
+		return connectorIds;
 	}
 
 	@Override
@@ -52,8 +68,11 @@ public long getLoadedOn() {
 		return loadedOn;
 	}
 
-	public void setConnectorId(String connectorId) {
-		this.connectorId = connectorId;
+	/**
+	 * @since 2.2
+     */
+	public void addConnectorId(String connectorId) {
+		this.connectorIds.add(connectorId);
 	}
 
 	public void setAttributes(RowAttribute... rowKeys) {

link-move/src/main/java/com/nhl/link/move/extractor/model/MutableExtractorModelContainer.java

@@ -2,7 +2,9 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.TreeMap;
 
 /**
@@ -12,14 +14,15 @@ public class MutableExtractorModelContainer implements ExtractorModelContainer {
 
 	private String location;
 	private String type;
-	private String connectorId;
+	private Set<String> connectorIds;
 	private long loadedOn;
 
 	private Map<String, ExtractorModel> extractors;
 
 	public MutableExtractorModelContainer(String location) {
 		this.location = location;
 
+		this.connectorIds = new HashSet<>();
 		// make sure the map is sorted for consistent iteration over
 		// extractors...
 		this.extractors = new TreeMap<>();
@@ -47,7 +50,18 @@ public String getType() {
 
 	@Override
 	public String getConnectorId() {
-		return connectorId;
+		if (connectorIds.isEmpty()) {
+			// connector IDs can be missing
+			return null;
+		} if (connectorIds.size() == 1) {
+			return connectorIds.iterator().next();
+		}
+		throw new IllegalStateException("Multiple connector IDs specified in model");
+	}
+
+	@Override
+	public Collection<String> getConnectorIds() {
+		return connectorIds;
 	}
 
 	@Override
@@ -60,8 +74,8 @@ public void addExtractor(String name, ExtractorModel extractor) {
 		extractors.put(name, extractor);
 	}
 
-	public void setConnectorId(String connectorId) {
-		this.connectorId = connectorId;
+	public void addConnectorId(String connectorId) {
+		this.connectorIds.add(connectorId);
 	}
 
 	public void setType(String type) {

link-move/src/main/java/com/nhl/link/move/extractor/parser/ExtractorModelParser_v1.java

@@ -75,7 +75,7 @@ protected void doParse(Element rootElement, MutableExtractorModelContainer conta
 				case "connectorId":
 					// enough to set this at the container level.. extractor
 					// will inherit the connectorId
-					container.setConnectorId(e.getTextContent());
+					container.addConnectorId(e.getTextContent());
 					break;
 				case "attributes":
 					processAttributes(e, extractor);

link-move/src/main/java/com/nhl/link/move/extractor/parser/ExtractorModelParser_v2.java

@@ -65,9 +65,12 @@ protected void parseContainer(MutableExtractorModelContainer container, NodeList
 					container.setType(e.getTextContent());
 					break;
 				case "connectorId":
-					container.setConnectorId(e.getTextContent());
+					container.addConnectorId(e.getTextContent());
 					break;
-
+				case "connectorIds": {
+					processConnectorIds(e, container);
+					break;
+				}
 				case "extractor":
 					parseExtractor(container, e);
 					break;
@@ -113,8 +116,12 @@ protected void parseExtractor(MutableExtractorModelContainer container, Element
 					extractor.setType(e.getTextContent());
 					break;
 				case "connectorId":
-					extractor.setConnectorId(e.getTextContent());
+					extractor.addConnectorId(e.getTextContent());
+					break;
+				case "connectorIds": {
+					processConnectorIds(e, extractor);
 					break;
+				}
 				case "attributes":
 					processAttributes(e, extractor);
 					break;
@@ -128,6 +135,36 @@ protected void parseExtractor(MutableExtractorModelContainer container, Element
 		container.addExtractor(extractor.getName(), new ContainerAwareExtractorModel(container, extractor));
 	}
 
+	private void processConnectorIds(Element connectorIds, MutableExtractorModel extractor) {
+		NodeList nodes = connectorIds.getChildNodes();
+		int len = nodes.getLength();
+		for (int i = 0; i < len; i++) {
+
+			Node c = nodes.item(i);
+			if (Node.ELEMENT_NODE == c.getNodeType()) {
+				Element e = (Element) c;
+				if ("id".equals(e.getNodeName())) {
+					extractor.addConnectorId(e.getTextContent());
+				}
+			}
+		}
+	}
+
+	private void processConnectorIds(Element connectorIds, MutableExtractorModelContainer container) {
+		NodeList nodes = connectorIds.getChildNodes();
+		int len = nodes.getLength();
+		for (int i = 0; i < len; i++) {
+
+			Node c = nodes.item(i);
+			if (Node.ELEMENT_NODE == c.getNodeType()) {
+				Element e = (Element) c;
+				if ("connectorId".equals(e.getNodeName())) {
+					container.addConnectorId(e.getTextContent());
+				}
+			}
+		}
+	}
+
 	protected void processAttributes(Element attributesNode, MutableExtractorModel extractor)
 			throws ClassNotFoundException, DOMException {
 

link-move/src/main/java/com/nhl/link/move/runtime/extractor/ReloadableExtractor.java

@@ -1,10 +1,14 @@
 package com.nhl.link.move.runtime.extractor;
 
+import java.util.Collection;
+import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import com.nhl.link.move.RowReader;
 import com.nhl.link.move.connect.Connector;
 import com.nhl.link.move.extractor.Extractor;
+import com.nhl.link.move.extractor.MultiExtractor;
 import com.nhl.link.move.extractor.model.ExtractorModel;
 import com.nhl.link.move.extractor.model.ExtractorName;
 import com.nhl.link.move.runtime.connect.IConnectorService;
@@ -64,7 +68,21 @@ private <T extends Connector> Extractor createExtractor(ExtractorModel model) {
 			throw new IllegalStateException("No factory mapped for Extractor type of '" + model.getType() + "'");
 		}
 
-		T connector = connectorService.getConnector(factory.getConnectorType(), model.getConnectorId());
+		Collection<String> connectorIds = model.getConnectorIds();
+		if (connectorIds.size() == 1) {
+			return createExtractor(model, factory, connectorIds.iterator().next());
+		} else {
+			List<Extractor> extractors = connectorIds.stream()
+					.map(id -> createExtractor(model, factory, id))
+					.collect(Collectors.toList());
+			return new MultiExtractor(extractors);
+		}
+	}
+
+	private <T extends Connector> Extractor createExtractor(ExtractorModel model,
+															IExtractorFactory<T> factory,
+															String connectorId) {
+		T connector = connectorService.getConnector(factory.getConnectorType(), connectorId);
 		return factory.createExtractor(connector, model);
 	}
 

link-move/src/main/resources/com/nhl/link/move/xsd/extractor_config_2.xsd

@@ -11,7 +11,6 @@
 
 	<element name="config" type="tns:ConfigType" />
 
-
 	<complexType name="ConfigType">
 
 		<annotation>
@@ -31,16 +30,10 @@
 				</annotation>
 			</element>
 
-			<element name="connectorId" type="string" minOccurs="1" maxOccurs="1">
-				<annotation>
-					<documentation source="description">
-						Defines a symbolic ID of the connector used by the extractor. Connector must
-						be
-						compatible with Extractor type specified above. ExtractorModelContainer's connectorId is inherited (and can be
-						overridden) by extractors.
-					</documentation>
-				</annotation>
-			</element>
+			<choice minOccurs="1" maxOccurs="1">
+				<element name="connectorId" type="tns:ConnectorIdType"/>
+				<element name="connectorIds" type="tns:ConnectorIdsType"/>
+			</choice>
 
 			<element name="extractor" type="tns:ExtractorType" minOccurs="0" maxOccurs="unbounded" />
 
@@ -67,13 +60,15 @@
 				</annotation>
 			</element>
 
-			<element name="connectorId" type="string" minOccurs="0" maxOccurs="1">
+			<choice minOccurs="0" maxOccurs="1">
 				<annotation>
 					<documentation source="description">
-						Overrides "connectorId" of the container
+						Overrides "connectorId"/"connectorIds" of the container
 					</documentation>
 				</annotation>
-			</element>
+				<element name="connectorId" type="tns:ConnectorIdType"/>
+				<element name="connectorIds" type="tns:ConnectorIdsType"/>
+			</choice>
 
 			<element name="attributes" minOccurs="0" maxOccurs="1">
 				<annotation>
@@ -123,5 +118,22 @@
 		</sequence>
 	</complexType>
 
+	<complexType name="ConnectorIdsType">
+		<sequence>
+			<element name="connectorId" type="tns:ConnectorIdType" minOccurs="1" maxOccurs="unbounded"/>
+		</sequence>
+	</complexType>
+
+	<simpleType name="ConnectorIdType">
+		<annotation>
+			<documentation source="description">
+				Defines a symbolic ID of the connector used by the extractor. Connector must
+				be
+				compatible with Extractor type specified above. ExtractorModelContainer's connectorId is inherited (and can be
+				overridden) by extractors.
+			</documentation>
+		</annotation>
+		<restriction base="string"/>
+	</simpleType>
 
 </schema>