/*
* Citrusleaf Aerospike Java Library
*
* Copyright 2009-2010 by Citrusleaf, Inc. All rights reserved.
*
* Availability of this source code to partners and customers includes
* redistribution rights covered by individual contract. Please check
* your contract for exact rights and responsibilities.
*/
package net.citrusleaf;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.concurrent.atomic.*;
import net.citrusleaf.CitrusleafClient.ClResult;
import net.citrusleaf.CitrusleafClient.SerializeException;
import net.citrusleaf.CLNode;
import net.citrusleaf.CitrusleafClient.ScanCallback;
import java.util.Date;
public class CLConnection {
/*
private static ConcurrentHashMap<String, ConcurrentLinkedQueue<CLConnection>> connHashMap =
new ConcurrentHashMap<String, ConcurrentLinkedQueue<CLConnection>>();
*/
// Need to stash address so I know which queue to put it back on
InetSocketAddress address;
Socket socket;
String connKey;
public java.util.Date lastUsed;
int timeout; // this is the default timeout value
int set_timeout; // this is what the timeout is currently set to, so as to avoid extra syscalls
public CLNode node; // just let external users set at will, don't bother with accessors
static AtomicInteger connCount = new AtomicInteger(0);
public CLConnection(InetSocketAddress address, String key ) throws UnknownHostException, IOException
{
this.address = address;
this.socket = null;
this.node = null;
this.lastUsed = new java.util.Date();
// set up conn key
if( key == null ) {
;
} else {
connKey = key;
}
this.timeout = CitrusleafClient.DEFAULT_TIMEOUT;
this.set_timeout = -1;
}
public CLConnection(InetSocketAddress address, String key, int timeout ) throws UnknownHostException, IOException {
this.address = address;
this.socket = null;
this.node = null;
this.lastUsed = new java.util.Date();
// set up conn key
if( key == null ){
setConnKey();
}else{
connKey = key;
}
this.timeout = timeout;
this.set_timeout = -1;
}
private void setConnKey()
{
this.connKey = this.address.toString();
}
public String getConnKey()
{
return connKey;
}
public void connect(int timeout) throws IOException {
socket = new Socket();
socket.connect(address, timeout);
connCount.getAndIncrement();
socket.setTcpNoDelay(true);
setTimeout(timeout);
node.validNetworkActivity();
// CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"connect: nconns "+connCount.get()+" "+socket.toString() );
}
public void connect() throws IOException {
this.connect(this.timeout);
}
public void close() {
if (socket == null) return;
try {
socket.close();
connCount.getAndDecrement();
} catch (IOException e) {
// e.printStackTrace();
;
}
socket = null;
// CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"connection close: nconns "+connCount.get() );
}
public void poolReturn(boolean connected) {
if (node != null) {
if (connected == true)
node.returnConnectionToPool(this);
else
node.closeConnection(this);
}
else if (socket != null) {
CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"pool return but no node, what's up with that");
try {
socket.close();
socket = null;
} catch (IOException e) {
;
}
}
// CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"connection: pool return: nconns "+connCount.get() );
}
// DEPRECATED [JOEY - 10/13/2011]
public void setTimeout(CitrusleafClient.ClOptions cl_opt ) throws IOException
{
// CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"set timeout from cl_opt");
try {
if (cl_opt == null) {
if (set_timeout != CitrusleafClient.DEFAULT_TIMEOUT) {
socket.setSoTimeout(CitrusleafClient.DEFAULT_TIMEOUT);
set_timeout = CitrusleafClient.DEFAULT_TIMEOUT;
// CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"set timeout from cl_opt - using default timeout "+socket.toString());
}
}
else {
if (set_timeout != cl_opt.timeout_ms) {
socket.setSoTimeout(cl_opt.timeout_ms);
set_timeout = cl_opt.timeout_ms;
// CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"set timeout from cl_opt - using "+timeout+" "+socket.toString());
}
}
}
catch (IOException e) {
// CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"set timeout threw exception");
if (node==null) node.networkFailure();
throw e;
}
}
public void setTimeout(int timeout ) throws IOException
{
if (timeout <= 0)
CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"set timeout with ZERO value probably a bad idea ");
// CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"set timeout with value "+timeout);
try {
if (this.set_timeout != timeout) {
socket.setSoTimeout(timeout);
this.set_timeout = timeout;
// CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"set timeout::: "+timeout+" "+socket.toString());
}
}
catch (IOException e) {
CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"set timeout threw exception");
if (node==null) node.networkFailure();
throw e;
}
}
public boolean isConnected()
{
if (socket == null) {
// CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO," CLConnection: socket is null for some reason");
return false;
}
return(socket.isConnected());
}
// so, it's better if the byte buffers are bigger
// every bytebuffer write actually does a "flush" implicity
// which bites the big one
public void send(CLBuffer clBuf) throws IOException
{
try {
OutputStream os = socket.getOutputStream();
os.write(clBuf.msg_write_buf, 0, (int)clBuf.msg_write_len);
}
catch (IOException e) {
CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"send throwing exception "+e.toString());
throw e;
}
this.lastUsed = new java.util.Date();
}
// This is a little peculiar, stylistically, but only the CL buffer
// knows the internals of how it's going to read something. In reality,
// it's going to "hard read" into a small buffer, then do a larger read
// for the rest if necessary. But that's something only it knows - the connection
// doesn't! but the connection has the input stream, so there.
public ClResult retrieve(CLBuffer clBuf) throws IOException, SerializeException
{
InputStream is;
ClResult rv = null;
try {
if (socket.getSoTimeout() < 1) {
CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"retrieve: bad SO timeout "+socket.getSoTimeout() );
} else if (socket.getSoTimeout() > 10000) {
CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"retrieve: bad SO timeout "+socket.getSoTimeout() );
}
is = socket.getInputStream();
if (clBuf.isReadIterate()) {
rv = clBuf.multi_get_and_parse(is);
} else if (clBuf.isBatchRead()) {
rv = clBuf.get_and_parse_batch(is);
} else {
rv = clBuf.get_and_parse(is);
}
}
catch (IOException e) {
CitrusleafClient.ClLog(CitrusleafClient.ClLogLevel.INFO,"retrieve throwing exception "+e.toString());
throw e;
}
this.lastUsed = new java.util.Date();
return ( rv );
}
public void reportNetworkFailure() {
if (node!=null) node.networkFailure();
}
public void reportNetworkSuccess() {
if (node != null) node.validNetworkActivity();
}
}