package org.pingles.cascading.cassandra;

import cascading.tap.SinkMode;
import cascading.tap.Tap;
import cascading.tap.hadoop.TapCollector;
import cascading.tap.hadoop.TapIterator;
import cascading.tuple.TupleEntryCollector;
import cascading.tuple.TupleEntryIterator;
import cascading.util.Util;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.pingles.cascading.cassandra.hadoop.ColumnFamilyInputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;

public class CassandraTap extends Tap {
    private static final Logger LOGGER =
        LoggerFactory.getLogger(CassandraScheme.class);
    private static final String URI_SCHEME = "cassandra";

    private final String initialAddress;
    private final Integer rpcPort;
    private String columnFamilyName;
    private String keyspace;

    public CassandraTap(
            String initialAddress, Integer rpcPort, String keyspace,
            String columnFamilyName, CassandraScheme scheme) {
        super(scheme, SinkMode.APPEND);
        this.initialAddress = initialAddress;
        this.rpcPort = rpcPort;
        this.columnFamilyName = columnFamilyName;
        this.keyspace = keyspace;
    }

    @Override
    public void sinkInit(JobConf conf) throws IOException {
        LOGGER.info("Created Cassandra tap {}", getPath());
        LOGGER.info("Sinking to column family: {}", columnFamilyName);

        super.sinkInit(conf);

        ConfigHelper.setOutputColumnFamily(conf, keyspace, columnFamilyName);
        ConfigHelper.setPartitioner(conf,
            "org.apache.cassandra.dht.RandomPartitioner");
        ConfigHelper.setInitialAddress(conf, this.initialAddress);
        ConfigHelper.setRpcPort(conf, this.rpcPort.toString());
    }

    @Override
    public void sourceInit(JobConf conf) throws IOException {
        LOGGER.info("Sourcing from column family: {}", columnFamilyName);

        FileInputFormat.addInputPaths(conf, getPath().toString());
        conf.setInputFormat(ColumnFamilyInputFormat.class);
        ConfigHelper.setInputColumnFamily(conf, keyspace, columnFamilyName);
        ConfigHelper.setInitialAddress(conf, initialAddress);
        ConfigHelper.setRpcPort(conf, rpcPort.toString());

        super.sourceInit(conf);
    }

    protected String getStringPath() {
        return String.format("%s://%s:%d/%s/%s",
            URI_SCHEME, initialAddress, rpcPort, keyspace, columnFamilyName);
    }

    @Override
    public Path getPath() {
        return new Path(getStringPath());
    }

    @Override
    public String toString() {
        return getClass().getSimpleName()
            + "[\"" + URI_SCHEME + "\"]"
            + "[\"" + Util.sanitizeUrl(getStringPath()) + "\"]";
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj) return true;
        if (obj == null || getClass() != obj.getClass()) return false;
        if (!super.equals(obj)) return false;

        CassandraTap tap = (CassandraTap) obj;
        if (!getScheme().equals(tap.getScheme())) return false;
        return getStringPath().equals(tap.getStringPath());
    }

    @Override
    public int hashCode() {
        int result = super.hashCode();
        result = 31 * result + getStringPath().hashCode();
        return result;
    }

    @Override
    public TupleEntryIterator openForRead(JobConf jobConf) throws IOException {
        return new TupleEntryIterator(
            getSourceFields(), new TapIterator(this, jobConf));
    }

    @Override
    public TupleEntryCollector openForWrite(JobConf jobConf) throws IOException {
        return new TapCollector(this, jobConf);
    }

    @Override
    public boolean makeDirs(JobConf jobConf) throws IOException {
        throw new UnsupportedOperationException(
            "makeDirs unsupported with Cassandra.");
    }

    @Override
    public boolean deletePath(JobConf jobConf) throws IOException {
        throw new UnsupportedOperationException(
            "deletePath unsupported with Cassandra.");
    }

    @Override
    public boolean pathExists(JobConf jobConf) throws IOException {
        throw new UnsupportedOperationException(
            "pathExists unsupported with Cassandra.");
    }

    @Override
    public long getPathModified(JobConf jobConf) throws IOException {
        throw new UnsupportedOperationException(
            "getPathModified unsupported with Cassandra.");
    }
}

@@ -819,150 +819,649 @@
 a%22;%0A
-%0A    private final String initialAddress;%0A    private final Integer rpcPort;%0A    private String columnFamilyName;%0A    private String keyspace;
+    private static final String THRIFT_PORT_KEY = %22cassandra.thrift.port%22;%0A    private static final String DEFAULT_RPC_PORT = %229160%22;%0A    private static final String DEFAULT_ADDRESS = %22localhost%22;%0A%0A    private final String initialAddress;%0A    private final Integer rpcPort;%0A    private String columnFamilyName;%0A    private String keyspace;%0A%0A    public CassandraTap(%0A            String keyspace, String columnFamilyName, CassandraScheme scheme) %7B%0A        super(scheme, SinkMode.APPEND);%0A        this.initialAddress = null;%0A        this.rpcPort = null;%0A        this.columnFamilyName = columnFamilyName;%0A        this.keyspace = keyspace;%0A    %7D%0A
 %0A%0A  
@@ -2252,128 +2252,25 @@
     
-ConfigHelper.setInitialAddress(conf, this.initialAddress);%0A        ConfigHelper.setRpcPort(conf, this.rpcPort.toString()
+endpointInit(conf
 );%0A 
@@ -2642,62 +2642,419 @@
     
-ConfigHelper.setInitialAddress(conf, initialAddress);%0A
+endpointInit(conf);%0A%0A        super.sourceInit(conf);%0A    %7D%0A%0A    protected void endpointInit(JobConf conf) throws IOException %7B%0A        if (initialAddress != null) %7B%0A            ConfigHelper.setInitialAddress(conf, initialAddress);%0A        %7D else if (ConfigHelper.getInitialAddress(conf) == null) %7B%0A            ConfigHelper.setInitialAddress(conf, DEFAULT_ADDRESS);%0A        %7D%0A%0A        if (rpcPort != null) %7B%0A    
     
@@ -3112,86 +3112,295 @@
 ));%0A
-%0A
+ 
        
- super.sourceInit(conf);%0A    %7D%0A%0A    protected String getStringPath() %7B
+%7D else if (conf.get(THRIFT_PORT_KEY) == null) %7B%0A            ConfigHelper.setRpcPort(conf, DEFAULT_RPC_PORT);%0A        %7D%0A    %7D%0A%0A    protected String getStringPath() %7B%0A        String host = (initialAddress != null)%0A            ? String.format(%22%25s:%25s%22, initialAddress, rpcPort) : %22%22;
 %0A   
@@ -3437,11 +3437,8 @@
 //%25s
-:%25d
 /%25s/
@@ -3466,38 +3466,19 @@
 SCHEME, 
-initialAddress, rpcPor
+hos
 t, keysp

Allow CassandraTap to use pre-configured defaults.