added test for installation and use of an external adaptor
diff --git a/asterix-external-data/src/main/resources/schema/library.xsd b/asterix-external-data/src/main/resources/schema/library.xsd
index f58175a..00f71f5 100644
--- a/asterix-external-data/src/main/resources/schema/library.xsd
+++ b/asterix-external-data/src/main/resources/schema/library.xsd
@@ -9,9 +9,7 @@
<xs:element name="return_type" type="xs:string" />
<xs:element name="function_type" type="xs:string" />
<xs:element name="definition" type="xs:string" />
-
<xs:element name="factory_class" type="xs:string" />
- <xs:element name="adaptor_type" type="xs:string" />
<!-- definition of complex elements -->
@@ -40,7 +38,6 @@
<xs:sequence>
<xs:element ref="lib:name" />
<xs:element ref="lib:factory_class" />
- <xs:element ref="lib:adaptor_type" />
</xs:sequence>
</xs:complexType>
</xs:element>
@@ -58,7 +55,7 @@
<xs:sequence>
<xs:element ref="lib:language" />
<xs:element ref="lib:libraryFunctions" minOccurs="0" />
- <xs:element ref="lib:libraryAdapters" minOccurs="0" />
+ <xs:element ref="lib:libraryAdapters" minOccurs="0" />
</xs:sequence>
</xs:complexType>
</xs:element>
diff --git a/asterix-external-data/src/test/java/edu/uci/ics/asterix/external/library/adaptor/TestTypedAdaptor.java b/asterix-external-data/src/test/java/edu/uci/ics/asterix/external/library/adaptor/TestTypedAdaptor.java
new file mode 100644
index 0000000..cb0142a
--- /dev/null
+++ b/asterix-external-data/src/test/java/edu/uci/ics/asterix/external/library/adaptor/TestTypedAdaptor.java
@@ -0,0 +1,134 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * 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 from
+ *
+ * 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 edu.uci.ics.asterix.external.library.adaptor;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import edu.uci.ics.asterix.external.dataset.adapter.StreamBasedAdapter;
+import edu.uci.ics.asterix.metadata.feeds.IFeedAdapter;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.hyracks.api.comm.IFrameWriter;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.dataflow.std.file.ITupleParserFactory;
+
+public class TestTypedAdaptor extends StreamBasedAdapter implements IFeedAdapter {
+
+ private static final long serialVersionUID = 1L;
+
+ private final PipedOutputStream pos;
+
+ private final PipedInputStream pis;
+
+ private final Map<String, String> configuration;
+
+ private DummyGenerator generator;
+
+ public TestTypedAdaptor(ITupleParserFactory parserFactory, ARecordType sourceDatatype, IHyracksTaskContext ctx,
+ Map<String, String> configuration) throws IOException {
+ super(parserFactory, sourceDatatype, ctx);
+ pos = new PipedOutputStream();
+ pis = new PipedInputStream(pos);
+ this.configuration = configuration;
+ }
+
+ @Override
+ public InputStream getInputStream(int partition) throws IOException {
+ return pis;
+ }
+
+ @Override
+ public void start(int partition, IFrameWriter frameWriter) throws Exception {
+ generator = new DummyGenerator(configuration, pos);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ executor.execute(generator);
+ super.start(partition, frameWriter);
+ }
+
+ private static class DummyGenerator implements Runnable {
+
+ private final int nOutputRecords;
+ private final OutputStream os;
+ private final byte[] EOL = "\n".getBytes();
+ private boolean continueIngestion;
+
+ public DummyGenerator(Map<String, String> configuration, OutputStream os) {
+ nOutputRecords = Integer.parseInt(configuration.get(TestTypedAdaptorFactory.KEY_NUM_OUTPUT_RECORDS));
+ this.os = os;
+ this.continueIngestion = true;
+ }
+
+ @Override
+ public void run() {
+ DummyRecord dummyRecord = new DummyRecord();
+ try {
+ int i = 0;
+ while (continueIngestion && i < nOutputRecords) {
+ dummyRecord.reset(i + 1, "" + (i + 1));
+ os.write(dummyRecord.toString().getBytes());
+ os.write(EOL);
+ i++;
+ }
+ } catch (IOException ioe) {
+ ioe.printStackTrace();
+ } finally {
+ try {
+ os.close();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ public void stop() {
+ continueIngestion = false;
+ }
+ }
+
+ private static class DummyRecord {
+
+ private int tweetid = 0;
+ private String text = null;
+
+ public void reset(int tweetid, String text) {
+ this.tweetid = tweetid;
+ this.text = text;
+ }
+
+ @Override
+ public String toString() {
+ return "{" + "\"tweetid\":" + "int64(" + "\"" + tweetid + "\"" + ")" + "," + "\"message-text\":" + "\""
+ + text + "\"" + "}";
+ }
+
+ }
+
+ @Override
+ public DataExchangeMode getDataExchangeMode() {
+ return DataExchangeMode.PUSH;
+ }
+
+ @Override
+ public void stop() throws Exception {
+ generator.stop();
+ }
+
+}
diff --git a/asterix-external-data/src/test/java/edu/uci/ics/asterix/external/library/adaptor/TestTypedAdaptorFactory.java b/asterix-external-data/src/test/java/edu/uci/ics/asterix/external/library/adaptor/TestTypedAdaptorFactory.java
new file mode 100644
index 0000000..39f7ab2
--- /dev/null
+++ b/asterix-external-data/src/test/java/edu/uci/ics/asterix/external/library/adaptor/TestTypedAdaptorFactory.java
@@ -0,0 +1,89 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * 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 from
+ *
+ * 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 edu.uci.ics.asterix.external.library.adaptor;
+
+import java.util.Map;
+
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
+import edu.uci.ics.asterix.metadata.feeds.ITypedAdapterFactory;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.BuiltinType;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.runtime.operators.file.AdmSchemafullRecordParserFactory;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksCountPartitionConstraint;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.dataflow.std.file.ITupleParserFactory;
+
+public class TestTypedAdaptorFactory implements ITypedAdapterFactory {
+
+ public static final String NAME = "test_typed_adaptor";
+
+ private static ARecordType adapterOutputType = initOutputType();
+
+ public static final String KEY_NUM_OUTPUT_RECORDS = "num_output_records";
+
+ private Map<String, String> configuration;
+
+ @Override
+ public SupportedOperation getSupportedOperations() {
+ return SupportedOperation.READ;
+ }
+
+ private static ARecordType initOutputType() {
+ String[] fieldNames = new String[] { "tweetid", "message-text" };
+ IAType[] fieldTypes = new IAType[] { BuiltinType.AINT64, BuiltinType.ASTRING };
+ ARecordType outputType = null;
+ try {
+ outputType = new ARecordType("TestTypedAdaptorOutputType", fieldNames, fieldTypes, false);
+ } catch (AsterixException exception) {
+ throw new IllegalStateException("Unable to create output type for adaptor " + NAME);
+ }
+ return outputType;
+ }
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public AdapterType getAdapterType() {
+ return AdapterType.TYPED;
+ }
+
+ @Override
+ public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
+ return new AlgebricksCountPartitionConstraint(1);
+ }
+
+ @Override
+ public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx, int partition) throws Exception {
+ ITupleParserFactory tupleParserFactory = new AdmSchemafullRecordParserFactory(adapterOutputType);
+ return new TestTypedAdaptor(tupleParserFactory, adapterOutputType, ctx, configuration);
+ }
+
+ @Override
+ public ARecordType getAdapterOutputType() {
+ return adapterOutputType;
+ }
+
+ @Override
+ public void configure(Map<String, String> configuration) throws Exception {
+ this.configuration = configuration;
+ }
+
+}
diff --git a/asterix-external-data/src/test/resources/text_functions.xml b/asterix-external-data/src/test/resources/text_functions.xml
index bd485a4..a0b7bf9 100644
--- a/asterix-external-data/src/test/resources/text_functions.xml
+++ b/asterix-external-data/src/test/resources/text_functions.xml
@@ -50,4 +50,10 @@
</definition>
</libraryFunction>
</libraryFunctions>
+ <libraryAdapters>
+ <libraryAdapter>
+ <name>test_typed_adaptor</name>
+ <factory_class>edu.uci.ics.asterix.external.library.adaptor.TestTypedAdaptorFactory</factory_class>
+ </libraryAdapter>
+ </libraryAdapters>
</externalLibrary>