package edu.uci.ics.asterix.external.adapter.factory;

import java.util.Map;

import edu.uci.ics.asterix.external.dataset.adapter.PullBasedAzureTwitterAdapter;
import edu.uci.ics.asterix.metadata.MetadataManager;
import edu.uci.ics.asterix.metadata.MetadataTransactionContext;
import edu.uci.ics.asterix.metadata.entities.Datatype;
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.ATypeTag;
import edu.uci.ics.asterix.om.types.IAType;
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;

public class PullBasedAzureTwitterAdapterFactory implements ITypedAdapterFactory {

    private static final long serialVersionUID = 1L;

    private static final String INGESTOR_CARDINALITY_KEY = "ingestor-cardinality";
    private static final String OUTPUT_TYPE_KEY = "output-type";

    private ARecordType recordType;
    private Map<String, String> configuration;

    @Override
    public SupportedOperation getSupportedOperations() {
        return SupportedOperation.READ;
    }

    @Override
    public String getName() {
        return "azure_twitter";
    }

    @Override
    public AdapterType getAdapterType() {
        return AdapterType.TYPED;
    }

    @Override
    public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
        String cardinalityStr = configuration.get(INGESTOR_CARDINALITY_KEY);
        int cardinality = cardinalityStr == null ? 1 : Integer.parseInt(cardinalityStr);
        return new AlgebricksCountPartitionConstraint(cardinality);
    }

    @Override
    public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx) throws Exception {
        return new PullBasedAzureTwitterAdapter(configuration, ctx, recordType);
    }

    @Override
    public ARecordType getAdapterOutputType() {
        return recordType;
    }

    @Override
    public void configure(Map<String, String> configuration) throws Exception {
        this.configuration = configuration;
        String fqOutputType = configuration.get(OUTPUT_TYPE_KEY);

        if (fqOutputType == null) {
            throw new IllegalArgumentException("No output type specified");
        }
        String[] dataverseAndType = fqOutputType.split("[.]");
        String dataverseName = dataverseAndType[0];
        String datatypeName = dataverseAndType[1];
        MetadataTransactionContext ctx = null;
        try {
            MetadataManager.INSTANCE.acquireReadLatch();
            ctx = MetadataManager.INSTANCE.beginTransaction();
            Datatype t = MetadataManager.INSTANCE.getDatatype(ctx, dataverseName, datatypeName);
            IAType type = t.getDatatype();
            if (type.getTypeTag() != ATypeTag.RECORD) {
                throw new IllegalStateException();
            }
            recordType = (ARecordType) t.getDatatype();
            MetadataManager.INSTANCE.commitTransaction(ctx);
        } catch (Exception e) {
            if (ctx != null) {
                MetadataManager.INSTANCE.abortTransaction(ctx);
            }
            throw e;
        } finally {
            MetadataManager.INSTANCE.releaseReadLatch();
        }
    }
}

@@ -649,37 +649,40 @@
 aints.Algebricks
-Count
+Absolute
 PartitionConstra
@@ -1010,35 +1010,33 @@
 ng INGESTOR_
-CARDINALITY
+LOCATIONS
 _KEY = %22inge
@@ -1040,27 +1040,25 @@
 ngestor-
-cardinality
+locations
 %22;%0A    p
@@ -1614,27 +1614,25 @@
  String 
-cardinality
+locations
 Str = co
@@ -1660,19 +1660,17 @@
 TOR_
-CARDINALITY
+LOCATIONS
 _KEY
@@ -1685,85 +1685,121 @@
    i
-nt cardinality = cardinalityStr == null ? 1 : Integer.parseInt(cardinalityStr
+f (locationsStr == null) %7B%0A            return null;%0A        %7D%0A        String%5B%5D locations = locationsStr.split(%22,%22
 );%0A 
@@ -1830,13 +1830,16 @@
 icks
-Count
+Absolute
 Part
@@ -1858,19 +1858,17 @@
 int(
-cardinality
+locations
 );%0A 

allow for absolute constraints or *all NCs* rather than count constraints