package nl.idgis.publisher.database;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
import java.util.List;

import nl.idgis.publisher.database.messages.Commit;
import nl.idgis.publisher.database.messages.NotFound;
import nl.idgis.publisher.database.messages.Query;
import nl.idgis.publisher.database.messages.Rollback;
import nl.idgis.publisher.database.messages.StreamingQuery;
import nl.idgis.publisher.protocol.messages.Ack;
import nl.idgis.publisher.protocol.messages.Failure;
import nl.idgis.publisher.stream.messages.NextItem;
import nl.idgis.publisher.utils.TypedIterable;
import nl.idgis.publisher.utils.TypedList;
import akka.actor.ActorRef;
import akka.actor.UntypedActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.japi.Function;

public abstract class JdbcTransaction extends UntypedActor {
	
	private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
	
	protected final Connection connection;
	
	private boolean answered = false;
	
	public JdbcTransaction(Connection connection) {
		this.connection = connection;
	}
	
	protected void executeQuery(Query query) throws Exception {
		unhandled(query);
	}
	
	protected void executeQuery(StreamingQuery query) throws Exception {
		unhandled(query);
	}
	
	@Override
	public void postStop() throws Exception {
		connection.close();
	};

	@Override
	public void onReceive(Object msg) throws Exception {
		if(msg instanceof Commit) {
			log.debug("committing transaction");
			
			connection.commit();
			
			getSender().tell(new Ack(), getSelf());
			getContext().stop(getSelf());
		} else if(msg instanceof Rollback) {
			log.debug("rolling back transaction");
			
			connection.rollback();
			
			getSender().tell(new Ack(), getSelf());
			getContext().stop(getSelf());
		} else if(msg instanceof Query) {
			try {				 
				executeQuery((Query)msg);				
				
				finish();
			} catch(SQLException e) {
				failure(e);
			}
		} else if(msg instanceof StreamingQuery) {
			try {				 
				executeQuery((StreamingQuery)msg);				
				
				finish();
			} catch(SQLException e) {
				failure(e);
			}
		} else {
			unhandled(msg);
		}
	}

	private void failure(SQLException e) throws SQLException {
		log.error(e, "query failure");
		
		getSender().tell(new Failure(e), getSelf());
		
		connection.close();
		getContext().stop(getSelf());
	}
	
public static class Prepared {
		
		PreparedStatement stmt;
		
		private Prepared(PreparedStatement stmt) {
			this.stmt = stmt;
		}
		
		public void execute(Object... args) throws Exception {
			execute(Arrays.asList(args));
		}		
		
		public void execute(List<Object> args) throws Exception {
			execute(args, new Function<Object, Object>() {

				@Override
				public Object apply(Object o) throws Exception {
					return o;
				}				
			});
		}
		
		public void execute(List<Object> args, Function<Object, Object> converter) throws Exception {
			int i = 1;
			
			for(Object arg : args) {
				stmt.setObject(i++, converter.apply(arg));
			}
			
			stmt.execute();
			stmt.close();
		}
	}
	
	public <T> void answer(Class<T> type, Iterable<T> msg) {
		answer(new TypedIterable<>(type, msg));
	}
	
	public void answer(TypedIterable<?> msg) {
		answer((Object)msg);
	}
	
	public <T> void answer(Class<T> type, List<T> msg) {
		answer(new TypedList<>(type, msg));
	}
	
	public void answer(TypedList<?> msg) {
		answer((Object)msg);
	}
	
	/**
	 * 
	 * @param msg
	 * 
	 * @deprecated use {@link #answer(Class, Iterable)} instead.
	 */
	@Deprecated
	public void answer(Iterable<?> msg) {
		answer((Object)msg);
	}
	
	private void answer() {
		if(answered) {
			throw new IllegalArgumentException("query already answered");
		}
		
		answered = true;
	}
	
	public void answerStreaming(ActorRef cursor) {
		answer();
		
		cursor.tell(new NextItem(), getSender());
	}
	
	public void answer(Object msg) {
		answer();
		
		Object response = msg == null ? new NotFound() : msg;
		
		getSender().tell(response, getSelf());		
	}
	
	public void execute(String sql) throws SQLException {
		Statement stmt = connection.createStatement();
		stmt.execute(sql);
		stmt.close();
	}
	
	public Prepared prepare(String sql) throws SQLException {
		return new Prepared(connection.prepareStatement(sql));
	}
	
	public void ack() {
		answer(new Ack());
	}
	
	void finish() {
		if(!answered) {
			throw new IllegalStateException("query not answered");
		}
		
		answered = false;
	}
}

@@ -1977,37 +1977,67 @@
 ery) %7B%0D%0A%09%09%09try %7B
+%0D%0A
 %09%09%09%09
- 
+log.debug(%22executing query%22);
 %0D%0A%09%09%09%09executeQue
@@ -2081,35 +2081,32 @@
 ();%0D%0A%09%09%09%7D catch(
-SQL
 Exception e) %7B%0D%0A
@@ -2187,12 +2187,53 @@
 y %7B%09
+%0D%0A
 %09%09%09
- 
+%09log.debug(%22executing streaming query%22);
 %0D%0A%09%09
@@ -2306,19 +2306,16 @@
 %7D catch(
-SQL
 Exceptio
@@ -2410,19 +2410,16 @@
 failure(
-SQL
 Exceptio
@@ -2536,35 +2536,8 @@
 );%0D%0A
-%09%09%0D%0A%09%09connection.close();%0D%0A
 %09%09ge
@@ -2572,16 +2572,17 @@
 %0A%09%7D%0D%0A%09%0D%0A
+%09
 public s
@@ -4701,32 +4701,68 @@
 ed%22);%0D%0A%09%09%7D%0D%0A%09%09%0D%0A
+%09%09log.debug(%22query answered%22);%0D%0A%09%09%0D%0A
 %09%09answered = fal

debug logging added