public class MD5Digest {
	static final char lookup[] = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f' };
	
	private MD5Digest() {
	}

	/**
	 * Encodes user/password/salt information in the following way: MD5(MD5(password + user) + salt)
	 * 
	 * @param aUser The connecting user.
	 * 
	 * @param aPassword The connecting user's password.
	 * 
	 * @param salt A four-salt sent by the server.
	 * 
	 * @return A 35-byte array, comprising the string "md5" and an MD5 digest.
	 */
	public static byte[] encode(String aUser, char[] aPassword, byte salt[]) {
		MessageDigest md;
		byte[] temp_digest, pass_digest;
		byte[] hex_digest = new byte[35];
		byte[] user = toUTF8(aUser);
		byte[] password = toUTF8(new String(aPassword));
		
		try {
			md = MessageDigest.getInstance("MD5");

			md.update(password);
			md.update(user);
			temp_digest = md.digest();

			bytesToHex(temp_digest, hex_digest, 0);
			md.update(hex_digest, 0, 32);
			md.update(salt);
			pass_digest = md.digest();

			bytesToHex(pass_digest, hex_digest, 3);
			hex_digest[0] = (byte) 'm';
			hex_digest[1] = (byte) 'd';
			hex_digest[2] = (byte) '5';
		} catch (Exception e) {
			; // "MessageDigest failure; " + e
		}

		return hex_digest;
	}
	
	private static byte[] toUTF8(String aString) {
		try {
			return aString.getBytes("UTF-8");
		} catch (UnsupportedEncodingException e) {
		}
		throw new IllegalArgumentException("UTF8 encoding not found");
	}

	/*
	 * Turn 16-byte stream into a human-readable 32-byte hex string
	 */
	private static void bytesToHex(byte[] bytes, byte[] hex, int offset) {
		int pos = offset;

		for (int i = 0; i < 16; i++) {
			int c = bytes[i] & 0xFF;
			int j = c >> 4;
			hex[pos++] = (byte) lookup[j];
			j = (c & 0xF);
			hex[pos++] = (byte) lookup[j];
		}
	}
} public class PromisedResult extends FulfillablePromise<Transaction> implements Handler<Transaction> {
	
	/** Call back method which delegates to the ResultHandler set up in the then method. */
	@Override
	public void handle(Transaction value) {
		fulfill(value);
	}
} public interface ResultEnd {

	/** Query has finished. count contains the number of rows returned */
	public abstract void end(int count, Transaction trx);

} public interface SingleResultFunction<T> {
	public abstract void result(T result, Transaction trx);
} public interface ResultListener extends ResultStart, ResultRow, ResultEnd {

} public interface ResultRow {

	/** Callback to deliver a row of data. Row also has a reference to the column model. Use this to store or pass on query results.
	 * Note that this method will not get called for queries that return 0 rows (duh!)
	 * @param row reference to the byte buffer that contains the data.
	 * @param trx the currently run transaction
	 */
	public abstract void row(Row row, Transaction trx);

} public interface ResultStart {

	/** Query was successful and results are incoming.
	 * Use this to prepare for the upcoming data by examining the column names and types */
	public abstract void start(Columns cols, Transaction trx);

} public abstract class SingleResult<T> implements ResultListener {
	T result;
	
	@Override
	public void start(Columns cols, Transaction trx) {
	}

	@SuppressWarnings("unchecked")
	@Override
	public void row(Row row, Transaction trx) {
		result = (T) row.asObject(0);
	}

	@Override
	public void end(int count, Transaction trx) {
		result(result, trx);
	}

	
	public abstract void result(T result, Transaction trx);

} public class AuthenticationRequest extends BackendMessage {

	private int authInfo() {
		return buffer.getInt(5);
	}
	
	/** Retrieve the salt if this is MD5 password authentication request. Will throw exception if you try to call this and needMD5Password returns false */
	public byte[] salt() {
		return buffer.getBytes(9, 13);
	}
	
	public boolean isAuthenticationOk() {
		return authInfo() == 0;
	}

	public boolean needCleartextPassword() {
		return authInfo() == 3;
	}
	
	public boolean needMD5Password() {
		return authInfo() == 5;
	}
	
	public boolean isAuthenticationSupported() {
		return needCleartextPassword() || needMD5Password();
	}
} public class BackendKeyData extends BackendMessage {

	public int getProcessID() {
		return buffer.getInt(5);
	}
	
	public int getSecretKey() {
		return buffer.getInt(9);
	}
} public class CommandComplete extends BackendMessage {

	public String getTag() {
		return readCString(5);
	}

} public class DataRow extends BackendMessage {
	// this is possibly a micro-optimization, as finding the data for a column is summing up a few ints
	// we are sacrificing heap for ease-of-use and cpu
	int[] indexes; // for each column, records pointer in buffer where the content is. Makes it easier to read column data which will most likely be done
	
	@Override 
	public void setBuffer(Buffer aBuffer) {
		super.setBuffer(aBuffer);
		short col = buffer.getShort(5);
		indexes = new int[col];
		for (int pos = 7,i = 0; i < col; i++) {
			indexes[i] = pos;
			int lenData = buffer.getInt(pos); // length is raw length of column data excluding the length information
			if (lenData == -1) { // NULL value
				pos += 4; // next column starts right away
			} else {
				pos += 4 + lenData;
			}
		}
	}
	
	/** Get a copy of the raw bytes making up the value of that column. Data will be copied for each call to this method.
	 * 
	 * @param column the column index (0-based)
	 * @return a buffer with the raw data
	 */
	public Buffer getRawBytes(int column) {
		int lenData = buffer.getInt(indexes[column]);
		if (lenData == -1) return new Buffer(0); // NULL
		return buffer.getBuffer(indexes[column] + 4,  indexes[column] + 4 + lenData);		
	}

	public Buffer getBuffer() {
		return buffer;
	}

	/** Return the length of the data in the column. If -1, there is no data, return null */
	public int len(int col) {
		return buffer.getInt(indexes[col]);
	}

	/** Return the position in the buffer where the data starts. If length is -1, there is no data, return null */
	public int pos(int col) {
		return indexes[col] + 4;
	}
	
} public class EmptyQueryResponse extends BackendMessage {

	public void read() {
		messageLength();
	}

} public class ErrorResponse extends BackendMessage {


	public String toString() {
		StringBuilder sb = new StringBuilder(messageLength());
		int pos = 5;
		
		byte fieldType = buffer.getByte(pos++);
		while (fieldType != 0 && pos < messageLength()) {
			sb.append((char) fieldType).append(':');
			pos = readCString(pos, sb);
			sb.append(' ');
			fieldType = buffer.getByte(pos++);
		}
		return sb.toString();
	}

	
} public class ParameterStatus extends BackendMessage {
	int pos;
	
	@Override
	public void setBuffer(Buffer aBuffer) {
		super.setBuffer(aBuffer);
		pos = 0;
	}

	public void addToMap(Map<String,String> params) {
		pos = 5;
		String key = nextCString();
		String value = nextCString();
		params.put(key, value);
	}

	private String nextCString() {
		StringBuilder sb = new StringBuilder();
		pos = readCString(pos, sb);
		return sb.toString();
	}
} public class ParseComplete extends BackendMessage {
	
} public class Password extends FrontendMessage {

	public Password(Buffer buffer) {
		super(buffer);
	}
	
	public Buffer write(char[] password) {
		int rec = startRecord('p');
		cString(password);
		stopRecord(rec);
		return buffer;
	}

	public Buffer write(byte[] encodedPassword) {
		int rec = startRecord('p');
		buffer.appendBytes(encodedPassword);
		zero();
		stopRecord(rec);
		return buffer;
	}

} public class Query extends FrontendMessage {

	public Query(Buffer buffer) {
		super(buffer);
	}

	public Buffer write(String sqlString) {
		int start = startRecord('Q');
		cString(sqlString);
		stopRecord(start);
		return buffer;
	}

} public class ReadyForQuery extends BackendMessage {
	
	public boolean isIdle() {
		return buffer.getByte(6) == 'I';
	}

} public class RowDescription extends BackendMessage {

	public short columnCount() {
		return buffer.getShort(5); // can be 0!
	}
	
	public Columns readColumns() {
		short count = columnCount();

		Columns cs = new Columns(columnCount());
		for (int pos = 7, i = 0; i < count; i++) {
			StringBuilder name = new StringBuilder();
			pos = readCString(pos, name);
			int tableID = buffer.getInt(pos);
			pos+=4;
			short colNumber = buffer.getShort(pos);
			pos+=2;
			int type = buffer.getInt(pos);
			pos+=4;
			short typeSize = buffer.getShort(pos);
			pos+=2;
			int typeModifier = buffer.getInt(pos);
			pos+=4;
			short formatCode = buffer.getShort(pos);
			pos+=2;
			
			cs.setColumn(i, name.toString(), type, formatCode);
		}
		return cs;
	}

} public class Startup extends FrontendMessage {
	
	public Startup(Buffer aBuffer) { super(aBuffer); };
	
	public Buffer write(String user, String database, String applicationName) {	
		int start = startRecord();
		int16(3);
		int16(0); // protocol version
		cString("user");
		cString(user);
		cString("database");
		cString(database);
		cString("application_name");
		cString(applicationName);
		zero();
		stopRecord(start);
		return buffer;
	}

} public class FulfillablePromise<V> implements Promise<V> {

  public static <V> FulfillablePromise<V> create() {
    return new FulfillablePromise<>();
  }

  private enum State {
    PENDING,
    FULFILLED,
    REJECTED,
  }

  private static abstract class Handler<V> {
    Handler<V> next;

    abstract void fulfill(V value);

    abstract void reject(Throwable reason);
  }

  private State state = State.PENDING;
  private V value;
  private Throwable reason;

  private Handler<V> first;
  private Handler<V> last;

  public  void fulfill(V value) {
    if (state != State.PENDING) {
      throw new IllegalStateException();
    }

    state = State.FULFILLED;
    this.value = value;

    for (Handler<V> handler = first; handler != null; handler = handler.next) {
      handler.fulfill(value);
    }
    first = last = null;
  }

  public  void reject(Throwable reason) {
    if (state != State.PENDING) {
      throw new IllegalStateException();
    }

    state = State.REJECTED;
    this.reason = reason;

    for (Handler<V> handler = first; handler != null; handler = handler.next) {
      handler.reject(reason);
    }
    first = last = null;
  }

  @Override
  public  <R> Promise<R> then(final Callback<? super V, R> callback) {
    switch (state) {
    case FULFILLED:
      try {
        return callback.onFulfilled(value);
      } catch (Throwable t) {
        return Promises.rejected(t);
      }
    case REJECTED:
      try {
        return callback.onRejected(reason);
      } catch (Throwable t) {
        return Promises.rejected(t);
      }
    case PENDING:
    default:
      requireNonNull(callback);
      final FulfillablePromise<R> promise = new FulfillablePromise<>();
      addHandler(new Handler<V>() {
        @Override
        void fulfill(V value) {
          try {
            chain(callback.onFulfilled(value), promise);
          } catch (Throwable t) {
            promise.reject(t);
          }
        }

        @Override
        void reject(Throwable reason) {
          try {
            chain(callback.onRejected(reason), promise);
          } catch (Throwable t) {
            promise.reject(t);
          }
        }

        private void chain(Promise<R> promise, final FulfillablePromise<R> into) {
          promise.done(new DoneCallback<R>() {
            @Override
            public void onFulfilled(R value) {
              into.fulfill(value);
            }

            @Override
            public void onRejected(Throwable reason) {
              into.reject(reason);
            }
          });
        }
      });
      return promise;
    }
  }

  @Override
  public <R> Promise<R> then(final ImmediateCallback<? super V, R> callback) {
    switch (state) {
    case FULFILLED:
      try {
        return Promises.fulfilled(callback.onFulfilled(value));
      } catch (Throwable t) {
        return Promises.rejected(t);
      }
    case REJECTED:
      try {
        return Promises.fulfilled(callback.onRejected(reason));
      } catch (Throwable t) {
        return Promises.rejected(t);
      }
    case PENDING:
    default:
      requireNonNull(callback);
      final FulfillablePromise<R> promise = new FulfillablePromise<>();
      addHandler(new Handler<V>() {
        @Override
        void fulfill(V value) {
          try {
            promise.fulfill(callback.onFulfilled(value));
          } catch (Throwable t) {
            promise.reject(t);
          }
        }

        @Override
        void reject(Throwable reason) {
          try {
            promise.fulfill(callback.onRejected(reason));
          } catch (Throwable t) {
            promise.reject(t);
          }
        }
      });
      return promise;
    }
  }

  @Override
  public  void done(final DoneCallback<? super V> callback) {
    switch (state) {
    case FULFILLED:
      callback.onFulfilled(value);
      break;
    case REJECTED:
        callback.onRejected(reason);
        break;
    case PENDING:
    default:
      requireNonNull(callback);
      addHandler(new Handler<V>() {
        @Override
        void fulfill(V value) {
          callback.onFulfilled(value);
        }

        @Override
        void reject(Throwable reason) {
          callback.onRejected(reason);
        }
      });
    }
  }

  @Override
  public void done() {
    switch (state) {
    case FULFILLED:
      // no-op
      break;
    case REJECTED:
      throw Promises.propagate(reason);
    case PENDING:
      addHandler(new Handler<V>() {
        @Override
        void fulfill(V value) {
          // no-op
        }

        @Override
        void reject(Throwable reason) {
          Promises.propagate(reason);
        }
      });
    }
  }

  private void addHandler(Handler<V> handler) {
    if (last == null) {
      first = last = handler;
    } else {
      last = last.next = handler;
    }
  }
} public interface Promise<V> {

  abstract class Callback<V, R> {
    public abstract Promise<R> onFulfilled(V value);

    public Promise<R> onRejected(Throwable reason) {
      return Promises.rejected(reason);
    }
  }

  abstract class ImmediateCallback<V, R> {
    public abstract R onFulfilled(V value) throws Throwable;

    public R onRejected(Throwable reason) throws Throwable {
      throw reason;
    }
  }

  abstract class DoneCallback<V> {
    public abstract void onFulfilled(V value);

    public void onRejected(Throwable reason) {
      Promises.propagate(reason);
    }
  }

  <R> Promise<R> then(Callback<? super V, R> callback);

  <R> Promise<R> then(ImmediateCallback<? super V, R> callback);

  void done(DoneCallback<? super V> callback);

  void done();
} abstract class DoneCallback<V> {
  public abstract void onFulfilled(V value);

  public void onRejected(Throwable reason) {
    Promises.propagate(reason);
  }
}
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.vertx.java.core.AsyncResult;
import org.vertx.java.core.AsyncResultHandler;
import org.vertx.java.core.Handler;
import org.vertx.java.core.buffer.Buffer;
import org.vertx.java.core.net.NetClient;
import org.vertx.java.core.net.NetSocket;
import com.sun.corba.se.impl.encoding.CodeSetConversion.BTCConverter;
import us.monoid.psql.async.auth.MD5Digest;
import us.monoid.psql.async.callback.PromisedResult;
import us.monoid.psql.async.callback.ResultEnd;
import us.monoid.psql.async.callback.SingleResultFunction;
import us.monoid.psql.async.callback.ResultListener;
import us.monoid.psql.async.callback.ResultRow;
import us.monoid.psql.async.callback.ResultStart;
import us.monoid.psql.async.callback.SingleResult;
import us.monoid.psql.async.message.AuthenticationRequest;
import us.monoid.psql.async.message.BackendKeyData;
import us.monoid.psql.async.message.CommandComplete;
import us.monoid.psql.async.message.DataRow;
import us.monoid.psql.async.message.EmptyQueryResponse;
import us.monoid.psql.async.message.ErrorResponse;
import us.monoid.psql.async.message.ParameterStatus;
import us.monoid.psql.async.message.ParseComplete;
import us.monoid.psql.async.message.Password;
import us.monoid.psql.async.message.Query;
import us.monoid.psql.async.message.ReadyForQuery;
import us.monoid.psql.async.message.RowDescription;
import us.monoid.psql.async.message.Startup;
import us.monoid.psql.async.promise.FulfillablePromise;
import us.monoid.psql.async.promise.Promise;
import us.monoid.psql.async.promise.Promise.DoneCallback;
package us.monoid.psql.async;





/**
 * A currently running transaction. In essence, a single connection to the Database which can be re-used as needed. Instances of this class are managed by the Postgres class
 * 
 * @author beders
 * 
 */
public class Transaction {
	static final Logger log = Logger.getLogger(Transaction.class.getName());

	Postgres pg;
	MessageParser mp = new MessageParser(this);
	Map<String, String> params = new HashMap<>();

	NetSocket socket;

	enum Phase {
		startup, ready, executing, queryResults, released, extended;
	}

	Phase phase = Phase.startup;

	Handler<Transaction> connectionHandler;
	Handler<Transaction> executionHandler;
	ResultStart resultStartListener;
	ResultEnd resultEndListener;
	ResultRow resultRowListener;

	private int processID; // process ID of the back-end
	private int secretKey; // secret used in cancel requests to the back-end

	private String lastResult;

	private Row currentRow; // current data transmission (columns, rows)

	private int rowCounter;

	public final DoneCallback<? super Transaction> release = new DoneCallback<Transaction>() {
		@Override
		public void onFulfilled(Transaction trx) {
			trx.release();
		}
	};

	Transaction(Postgres aPostgresDB) {
		pg = aPostgresDB;
	}

	void connect(Handler<Transaction> aConnectionHandler) {
		connectionHandler = aConnectionHandler;
		NetClient client = pg.vertx.createNetClient();
		// client.setReceiveBufferSize(1024);
		client.connect(pg.port, pg.host, new AsyncResultHandler<NetSocket>() {
			@Override
			public void handle(AsyncResult<NetSocket> event) {
				if (event.succeeded()) {
					socket = event.result();
					socket.dataHandler(new Handler<Buffer>() {
						@Override
						public void handle(Buffer buffer) {
							mp.parseMessages(buffer,0,buffer.length());
						}
					});
					socket.write(new Startup(new Buffer(20)).write(pg.user, pg.db, pg.applicationName + "-" + this.hashCode()));
				} else {
					connectionHandler.handle(null);
				}
			}
		});
	}


	protected void dispatch(Buffer buffer) {
		if (log.isLoggable(Level.FINEST)) log.finest(debugMessage(buffer));
		pg.dispatch(buffer, this);
	}

	void on(CommandComplete commandComplete) {
		if (executionHandler != null) {
			this.lastResult = commandComplete.getTag();
			phase = Phase.ready;
			executionHandler.handle(this);
		}
	}

	/** Start up phase. Figuring out how to authenticate */
	void on(AuthenticationRequest ar) {
		if (ar.isAuthenticationOk()) {
			this.lastResult = "AUTHENTICATED";
			connectionHandler.handle(this);
		} else if (ar.isAuthenticationSupported()) {
			if (ar.needCleartextPassword()) {
				socket.write(new Password(new Buffer(20)).write(pg.password)); // send PasswordMessage --> onCheckPasswordMessage