package nl.idgis.publisher.provider.database;

import java.sql.Date;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;

import oracle.sql.STRUCT;

import org.deegree.geometry.io.WKBWriter;
import org.deegree.sqldialect.oracle.sdo.SDOGeometryConverter;

import nl.idgis.publisher.provider.database.messages.FetchTable;
import nl.idgis.publisher.provider.protocol.Record;
import nl.idgis.publisher.provider.protocol.Records;
import nl.idgis.publisher.provider.protocol.UnsupportedType;
import nl.idgis.publisher.provider.protocol.WKBGeometry;
import nl.idgis.publisher.stream.StreamCursor;

import akka.actor.Props;
import akka.event.Logging;
import akka.event.LoggingAdapter;

@SuppressWarnings("deprecation")
public class DatabaseCursor extends StreamCursor<ResultSet, Records> {
	
	private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
	
	private final SDOGeometryConverter converter = new SDOGeometryConverter();
	
	private final FetchTable fetchTable;
	
	private final ExecutorService executorService;

	public DatabaseCursor(ResultSet t, FetchTable fetchTable, ExecutorService executorService) {
		super(t);
		
		this.fetchTable = fetchTable;
		this.executorService = executorService;
	}
	
	public static Props props(ResultSet t, FetchTable fetchTable, ExecutorService executorService) {
		return Props.create(DatabaseCursor.class, t, fetchTable, executorService);
	}
	
	private Object convert(Object value) throws Exception {
		if(value == null 
				|| value instanceof String
				|| value instanceof Number
				|| value instanceof Date
				|| value instanceof Timestamp) {
			
			return value;
		} else if(value instanceof STRUCT) {
			return new WKBGeometry(WKBWriter.write(converter.toGeometry((STRUCT) value, null)));
		}
		
		return new UnsupportedType(value.getClass().getCanonicalName());
	}
	
	private Record toRecord() throws Exception {
		int columnCount = t.getMetaData().getColumnCount();
		
		List<Object> values = new ArrayList<>();
		for(int j = 0; j < columnCount; j++) {
			Object o = t.getObject(j + 1);
			values.add(convert(o));
		}
		
		return new Record(values);
	}

	@Override
	protected CompletableFuture<Records> next() {
		log.debug("fetching next records");
		
		CompletableFuture<Records> future = new CompletableFuture<>();
		
		int messageSize = fetchTable.getMessageSize();
		executorService.execute(() -> {
			try {
				List<Record> records = new ArrayList<>();
				records.add(toRecord());
				
				for(int i = 1; i < messageSize; i++) {
					if(!t.next()) {
						break;
					}
					
					records.add(toRecord());
				}
				
				future.complete(new Records(records));
			} catch(Throwable t) {
				log.error("failed to fetch records: {}", t);
				
				future.completeExceptionally(t);
			}
		});
		
		return future;
	}

	@Override
	protected boolean hasNext() throws Exception {		
		return t.next();
	}
	
	@Override
	public void postStop() throws SQLException {
		t.getStatement().close();
		t.close();
	}
}

@@ -46,31 +46,8 @@
 %0D%0A%0D%0A
-import java.sql.Date;%0D%0A
 impo
@@ -105,36 +105,8 @@
 n;%0D%0A
-import java.sql.Timestamp;%0D%0A
 impo
@@ -384,24 +384,72 @@
 nverter;%0D%0A%0D%0A
+import nl.idgis.publisher.domain.service.Type;%0D%0A
 import nl.id
@@ -498,24 +498,81 @@
 etchTable;%0D%0A
+import nl.idgis.publisher.provider.protocol.ColumnInfo;%0D%0A
 import nl.id
@@ -670,70 +670,8 @@
 s;%0D%0A
-import nl.idgis.publisher.provider.protocol.UnsupportedType;%0D%0A
 impo
@@ -1638,16 +1638,39 @@
 convert(
+ColumnInfo columnInfo, 
 Object v
@@ -1704,214 +1704,45 @@
 %09if(
-value == null %0D%0A%09%09%09%09%7C%7C value instanceof String%0D%0A%09%09%09%09%7C%7C value instanceof Number%0D%0A%09%09%09%09%7C%7C value instanceof Date%0D%0A%09%09%09%09%7C%7C value instanceof Timestamp) %7B%0D%0A%09%09%09%0D%0A%09%09%09return value;%0D%0A%09%09%7D else if(value instanceof STRUCT
+columnInfo.getType() == Type.GEOMETRY
 ) %7B%0D
@@ -1830,39 +1830,43 @@
 ull)));%0D%0A%09%09%7D
+ else %7B
 %0D%0A%09%09
-%0D%0A%09
 %09return 
 new Unsuppor
@@ -1857,65 +1857,19 @@
 urn 
-new UnsupportedType(value.getClass().getCanonicalName());
+value;%0D%0A%09%09%7D
 %0D%0A%09%7D
@@ -1924,67 +1924,8 @@
  %7B%0D%0A
-%09%09int columnCount = t.getMetaData().getColumnCount();%0D%0A%09%09%0D%0A
 %09%09Li
@@ -1970,20 +1970,20 @@
 %0D%0A%09%09
-for(
+%0D%0A%09%09
 int j = 
 0; j
@@ -1982,31 +1982,69 @@
 j = 
-0; j %3C columnCount; j++
+1;%0D%0A%09%09for(ColumnInfo columnInfo : fetchTable.getColumnInfos()
 ) %7B%0D
@@ -2054,17 +2054,21 @@
 %09Object 
-o
+value
  = t.get
@@ -2079,12 +2079,10 @@
 ct(j
- + 1
+++
 );%0D%0A
@@ -2103,17 +2103,33 @@
 convert(
-o
+columnInfo, value
 ));%0D%0A%09%09%7D

use ColumnInfo to determine what to do which a column value