Lots of changes

pom.xml

@@ -19,7 +19,7 @@
 		<dependency>
 			<groupId>io.teknek</groupId>
 			<artifactId>nit-compiler</artifactId>
-			<version>0.0.2</version>
+			<version>0.0.3-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>junit</groupId>
@@ -30,7 +30,7 @@
 		<dependency>
 			<groupId>org.apache.cassandra</groupId>
 			<artifactId>cassandra-all</artifactId>
-			<version>1.2.3</version>
+			<version>1.2.16</version>
 			<scope>provided</scope>
 		</dependency>
 		<dependency>
@@ -68,7 +68,14 @@
             <artifactId>mockito-all</artifactId>
             <version>1.9.0</version>
         </dependency>
+                        <dependency>
+        <groupId>org.mortbay.jetty</groupId>
+        <artifactId>jetty</artifactId>
+        <version>7.0.0.pre5</version>
+</dependency>
+
     </dependencies>
+
 	<build>
 		<plugins>
 			<plugin>

src/main/java/io/teknek/intravert/action/filter/Filter.java

@@ -0,0 +1,7 @@
+package io.teknek.intravert.action.filter;
+
+import java.util.Map;
+
+public interface Filter {
+  Map filter(Map m);
+}

src/main/java/io/teknek/intravert/action/filter/FilterFactory.java

@@ -0,0 +1,23 @@
+package io.teknek.intravert.action.filter;
+
+import java.util.Map;
+
+import groovy.lang.Closure;
+
+public class FilterFactory {
+
+  public static Filter createFilter(final Object object){
+    if (object instanceof Filter){
+      return (Filter) object;
+    } else if (object instanceof Closure){
+      return new Filter(){
+        public Map filter(Map m) {
+          return ((Closure<Map>)object).call(m);
+        }
+      };
+    } else {
+      throw new RuntimeException("Do not know what to do with "+ object);
+    }
+  }
+
+}

src/main/java/io/teknek/intravert/action/impl/CreateFilterAction.java

@@ -0,0 +1,30 @@
+package io.teknek.intravert.action.impl;
+
+import io.teknek.intravert.action.Action;
+import io.teknek.intravert.action.filter.Filter;
+import io.teknek.intravert.model.Operation;
+import io.teknek.intravert.model.Response;
+import io.teknek.intravert.service.ApplicationContext;
+import io.teknek.intravert.service.RequestContext;
+import io.teknek.nit.NitDesc;
+import io.teknek.nit.NitException;
+
+public class CreateFilterAction implements Action {
+
+  @Override
+  public void doAction(Operation operation, Response response, RequestContext request,
+          ApplicationContext application) {
+    String name = (String) operation.getArguments().get("name");
+    NitDesc n = new NitDesc();
+    n.setSpec(NitDesc.NitSpec.valueOf((String) operation.getArguments().get("spec")));
+    n.setTheClass((String) operation.getArguments().get("theClass"));
+    n.setScript((String) operation.getArguments().get("script"));
+    try {
+      Filter f = io.teknek.nit.NitFactory.construct(n);
+      request.getSession().putFilter(name, f);
+    } catch (NitException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+}

src/main/java/io/teknek/intravert/action/impl/CreateKeyspaceAction.java

@@ -0,0 +1,46 @@
+package io.teknek.intravert.action.impl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.KSMetaData;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.service.MigrationManager;
+import org.apache.cassandra.thrift.KsDef;
+
+import io.teknek.intravert.action.Action;
+import io.teknek.intravert.model.Operation;
+import io.teknek.intravert.model.Response;
+import io.teknek.intravert.service.ApplicationContext;
+import io.teknek.intravert.service.RequestContext;
+
+public class CreateKeyspaceAction implements Action {
+
+  @Override
+  public void doAction(Operation operation, Response response, RequestContext request,
+          ApplicationContext application) {
+    String keyspace = (String) operation.getArguments().get("name");
+    int replication = (Integer) operation.getArguments().get("replication");
+
+
+    Collection<CFMetaData> cfDefs = new ArrayList<CFMetaData>(0);
+    KsDef def = new KsDef();
+    def.setName(keyspace);
+    def.setStrategy_class("SimpleStrategy");
+    Map<String, String> strat = new HashMap<String, String>();
+    strat.put("replication_factor", Integer.toString(replication));
+    def.setStrategy_options(strat);
+    KSMetaData ksm = null;
+    try {
+        ksm = KSMetaData.fromThrift(def,
+            cfDefs.toArray(new CFMetaData[cfDefs.size()]));
+        MigrationManager.announceNewKeyspace(ksm);
+    } catch (ConfigurationException e) {
+        throw new RuntimeException(e);
+    }
+  }
+
+}

src/main/java/io/teknek/intravert/daemon/IntravertCassandraServer.java

@@ -0,0 +1,72 @@
+package io.teknek.intravert.daemon;
+import io.teknek.intravert.service.DefaultIntravertService;
+import io.teknek.intravert.service.IntravertService;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletResponse;
+import javax.servlet.http.HttpServletRequest;
+
+import org.apache.cassandra.service.CassandraDaemon.Server;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.mortbay.jetty.Handler;
+import org.mortbay.jetty.HttpConnection;
+import org.mortbay.jetty.Request;
+import org.mortbay.jetty.handler.AbstractHandler;
+
+public class IntravertCassandraServer implements Server {
+
+  private static final AtomicBoolean RUNNING = new AtomicBoolean(false);
+  private org.mortbay.jetty.Server server;
+  public static final int port = 7654;
+  private IntravertService intraService;
+  private static ObjectMapper MAPPER = new ObjectMapper();
+
+  @Override
+  public boolean isRunning() {
+    return RUNNING.get();
+  }
+
+  @Override
+  public void start() {
+    intraService = new DefaultIntravertService();
+    server = new org.mortbay.jetty.Server(port);
+    server.setHandler(getHandler());
+    try {
+      server.start();
+      RUNNING.set(true);
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public void stop() {
+    try {
+      server.stop();
+      RUNNING.set(false);
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }    
+  }
+
+  private Handler getHandler(){
+    final IntravertService copy = this.intraService;
+    AbstractHandler handler = new AbstractHandler() {
+      public void handle(String target, HttpServletRequest request, HttpServletResponse response,
+              int dispatch) throws IOException, ServletException {
+        Request baseRequest = request instanceof Request ? (Request) request : HttpConnection
+                .getCurrentConnection().getRequest();
+        String url = baseRequest.getRequestURI();
+        io.teknek.intravert.model.Request requestFromBody = MAPPER.readValue(baseRequest.getInputStream(), io.teknek.intravert.model.Request.class);
+        response.setStatus(HttpServletResponse.SC_OK);        
+        baseRequest.setHandled(true);
+        MAPPER.writeValue(response.getOutputStream(), copy.doRequest(requestFromBody));
+      }
+    };
+    return handler;
+
+  }
+}

src/main/java/io/teknek/intravert/model/Response.java

@@ -11,6 +11,7 @@ public class Response {
 
   public Response(){
     results = new HashMap<String,Object>();
+    metaData = new HashMap<String,Object>();
   }
 
   public String getExceptionMessage() {

src/main/java/io/teknek/intravert/service/Session.java

@@ -1,8 +1,18 @@
 package io.teknek.intravert.service;
 
+import io.teknek.intravert.action.filter.Filter;
+
+import java.util.HashMap;
+import java.util.Map;
+
 public class Session {
 
   private String currentKeyspace;
+  private Map<String,Filter> filters;
+
+  public Session(){
+    filters = new HashMap<>();
+  }
 
   public String getCurrentKeyspace() {
     return currentKeyspace;
@@ -12,4 +22,15 @@ public void setCurrentKeyspace(String currentKeyspace) {
     this.currentKeyspace = currentKeyspace;
   }
 
+  public void putFilter(String name, Filter f){
+    filters.put(name, f);
+  }
+
+  public Filter getFilter(String name){
+    Filter f = filters.get(name);
+    if (f == null){
+      throw new RuntimeException ("filter not found");
+    }
+    return f;
+  }
 }