package org.ardverk.gibson.appender;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

import org.ardverk.gibson.core.DatastoreFactory;
import org.ardverk.gibson.core.Event;

import com.google.code.morphia.Datastore;
import com.mongodb.Mongo;
import com.mongodb.MongoURI;

class MongoTransport implements Transport {
  
  private static final ExecutorService EXECUTOR = Executors.newCachedThreadPool(
      new TransportThreadFactory(MongoTransport.class));
  
  private List<Event> queue = new ArrayList<Event>();
  
  private final Console console;
  
  private final String database;
  
  private boolean open = true;
  
  private boolean connected = false;
  
  private Future<?> future = null;
  
  public MongoTransport(Console console, String database) {
    this.console = console;
    this.database = database;
  }

  @Override
  public synchronized boolean isConnected() {
    return connected;
  }

  @Override
  public synchronized void connect(MongoURI uri) throws IOException {
    if (!open) {
      throw new IOException("closed");
    }
    
    if (connected) {
      throw new IOException("connected");
    }
    
    if (console.isInfoEnabled()) {
      console.info("Connecting to: " + uri);
    }
    
    Mongo mongo = new Mongo(uri);
    
    DatastoreFactory factory = new DatastoreFactory(mongo);
    final Datastore datastore = factory.createDatastore(database);
    
    Runnable task = new Runnable() {
      @Override
      public void run() {
        try {
          process(datastore);
        } catch (InterruptedException err) {
          console.error("InterruptedException", err);
        } finally {
          destroy(datastore);
        }
      }
    };
    
    this.future = EXECUTOR.submit(task);
    this.connected = true;
  }
  
  private void destroy(Datastore datastore) {
    try {
      close();
    } finally {
      datastore.getMongo().close();
    }
  }
  
  @Override
  public synchronized void close() {
    if (open) {
      open = false;
      connected = false;
      
      if (future != null) {
        future.cancel(true);
      }
      
      notifyAll();
    }
  }
  
  private void process(Datastore datastore) throws InterruptedException {
    
    while (true) {
      
      List<Event> events = null;
      
      synchronized (this) {
        while (open && queue.isEmpty()) {
          wait();
        }
        
        if (!open) {
          break;
        }
        
        events = queue;
        queue = new ArrayList<Event>();
      }
      
      try {
        datastore.save(events);
      } catch (Exception err) {
        console.error("Exception", err);
      }
    }
  }
  
  @Override
  public void send(Event event) {
    if (event == null) {
      throw new NullPointerException("event");
    }
    
    synchronized (this) {
      if (open && connected) {
        queue.add(event);
        notifyAll();
      }
    }
  }
  
  private static class TransportThreadFactory implements ThreadFactory {

    private final AtomicInteger counter = new AtomicInteger();
    
    private final String name;
    
    public TransportThreadFactory(Class<?> clazz) {
      this.name = clazz.getSimpleName() + "Thread";
    }
    
    @Override
    public Thread newThread(Runnable r) {
      Thread thread = new Thread(r, name + "-" + counter.getAndIncrement());
      thread.setDaemon(true);
      return thread;
    }
  }
}

@@ -450,24 +450,59 @@
 .Datastore;%0A
+import com.mongodb.DBTCPConnector;%0A
 import com.m
@@ -515,16 +515,16 @@
 .Mongo;%0A
-
 import c
@@ -932,24 +932,57 @@
 = false;%0A  %0A
+  private Mongo mongo = null;%0A  %0A
   private Fu
@@ -1196,24 +1196,243 @@
     
-return connected
+if (!connected) %7B%0A      return false;%0A    %7D%0A    %0A    if (mongo == null) %7B%0A      return false;%0A    %7D%0A    %0A    DBTCPConnector connector = mongo.getConnector();%0A    if (!connector.isOpen()) %7B%0A      return false;%0A    %7D%0A    %0A    return true
 ;%0A  
@@ -1752,22 +1752,21 @@
    %0A    
-Mongo 
+this.
 mongo = 
@@ -2178,25 +2178,21 @@
 destroy(
-datastore
+mongo
 );%0A     
@@ -2314,35 +2314,27 @@
 destroy(
-Datastore datastore
+Mongo mongo
 ) %7B%0A    
@@ -2380,28 +2380,13 @@
     
-datastore.getM
+m
 ongo
-()
 .clo
@@ -3324,25 +3324,27 @@
 open && 
-c
+isC
 onnected
 ) %7B%0A    
@@ -3327,32 +3327,34 @@
 n && isConnected
+()
 ) %7B%0A        queu

Make sure we're actually connected to MongoDB (takes care of network partitions)