Cayenne Extractor

link-move/src/main/java/com/nhl/link/move/runtime/cayenne/CayenneConnector.java

@@ -0,0 +1,33 @@
+package com.nhl.link.move.runtime.cayenne;
+
+import com.nhl.link.move.runtime.jdbc.JdbcConnector;
+import org.apache.cayenne.ObjectContext;
+import org.apache.cayenne.configuration.server.ServerRuntime;
+import org.apache.cayenne.di.BeforeScopeEnd;
+
+/**
+ * A {@link JdbcConnector} based on externally provided Cayenne's ServerRuntime.
+ * 
+ * @since 3.0
+ */
+public class CayenneConnector implements JdbcConnector {
+
+	private final ServerRuntime runtime;
+	private final ObjectContext sharedContext;
+
+	public CayenneConnector(ServerRuntime runtime) {
+		this.runtime = runtime;
+		this.sharedContext = runtime.newContext();
+	}
+
+	@Override
+	public ObjectContext sharedContext() {
+		return sharedContext;
+	}
+
+	@BeforeScopeEnd
+	@Override
+	public void shutdown() {
+		runtime.shutdown();
+	}
+}

link-move/src/main/java/com/nhl/link/move/runtime/cayenne/CayenneExtractor.java

@@ -0,0 +1,66 @@
+package com.nhl.link.move.runtime.cayenne;
+
+import com.nhl.link.move.BaseRowAttribute;
+import com.nhl.link.move.Execution;
+import com.nhl.link.move.RowAttribute;
+import com.nhl.link.move.RowReader;
+import com.nhl.link.move.extractor.Extractor;
+import com.nhl.link.move.runtime.jdbc.DataRowIterator;
+import com.nhl.link.move.runtime.jdbc.JdbcRowReader;
+import org.apache.cayenne.DataRow;
+import org.apache.cayenne.ObjectContext;
+import org.apache.cayenne.exp.parser.ASTDbPath;
+import org.apache.cayenne.map.DbAttribute;
+import org.apache.cayenne.map.DbEntity;
+import org.apache.cayenne.query.ObjectSelect;
+
+import java.util.Objects;
+
+/**
+ * Extractor that uses Cayenne to stream data row-by-row from the specified database entity.
+ *
+ * @since 3.0
+ */
+public class CayenneExtractor implements Extractor {
+
+    private final ObjectContext context;
+    private final RowAttribute[] rowHeader;
+    private final String objEntityName;
+
+    public CayenneExtractor(ObjectContext context, RowAttribute[] rowHeader, String objEntityName) {
+        this.context = context;
+        this.rowHeader = rowHeader;
+        this.objEntityName = Objects.requireNonNull(objEntityName);
+    }
+
+    @Override
+    public RowReader getReader(Execution exec) {
+        DbEntity sourceEntity = context.getEntityResolver().getObjEntity(objEntityName).getDbEntity();
+        ObjectSelect<DataRow> select = ObjectSelect.dbQuery(sourceEntity.getName());
+
+        DataRowIterator it = new DataRowIterator(context.iterator(select));
+        RowAttribute[] rowHeader;
+        if (this.rowHeader != null) {
+            rowHeader = this.rowHeader;
+        } else {
+            rowHeader = new RowAttribute[sourceEntity.getAttributes().size()];
+            int ordinal = 0;
+            for (DbAttribute attr : sourceEntity.getAttributes()) {
+                rowHeader[ordinal] = new BaseRowAttribute(
+                        null,
+                        attr.getName(),
+                        convertToDbPath(attr.getName()),
+                        ordinal
+                );
+                ordinal++;
+            }
+        }
+
+        exec.getLogger().extractorStarted(rowHeader, select);
+        return new JdbcRowReader(rowHeader, it);
+    }
+
+    private static String convertToDbPath(String path) {
+        return ASTDbPath.DB_PREFIX + path;
+    }
+}

link-move/src/main/java/com/nhl/link/move/runtime/cayenne/CayenneExtractorFactory.java

@@ -0,0 +1,32 @@
+package com.nhl.link.move.runtime.cayenne;
+
+import com.nhl.link.move.extractor.Extractor;
+import com.nhl.link.move.extractor.model.ExtractorModel;
+import com.nhl.link.move.runtime.extractor.IExtractorFactory;
+
+/**
+ * Factory for {@link CayenneExtractor}
+ *
+ * @since 3.0
+ */
+public class CayenneExtractorFactory implements IExtractorFactory<CayenneConnector> {
+
+    private static final String CAYENNE_EXTRACTOR_TYPE = "cayenne";
+    private static final String SOURCE_ENTITY_PROPERTY = "extractor.cayenne.entity";
+
+    @Override
+    public String getExtractorType() {
+        return CAYENNE_EXTRACTOR_TYPE;
+    }
+
+    @Override
+    public Class<CayenneConnector> getConnectorType() {
+        return CayenneConnector.class;
+    }
+
+    @Override
+    public Extractor createExtractor(CayenneConnector connector, ExtractorModel model) {
+        String sourceEntity = model.getPropertyValue(SOURCE_ENTITY_PROPERTY);
+        return new CayenneExtractor(connector.sharedContext(), model.getAttributes(), sourceEntity);
+    }
+}

link-move/src/main/resources/META-INF/services/com.nhl.link.move.runtime.extractor.IExtractorFactory

@@ -1,2 +1,3 @@
 com.nhl.link.move.runtime.jdbc.JdbcExtractorFactory
-com.nhl.link.move.runtime.xml.XmlExtractorFactory
+com.nhl.link.move.runtime.cayenne.CayenneExtractorFactory
+com.nhl.link.move.runtime.xml.XmlExtractorFactory