plugins/org.eclipse.actf.util.httpproxy/src/org/eclipse/actf/util/internal/httpproxy/core/ClientConnection.java - actf/org.eclipse.actf.common - Git at Google

 /*******************************************************************************
  * Copyright (c) 2007 IBM Corporation and Others
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
  * which accompanies this distribution, and is available at
  * http://www.eclipse.org/legal/epl-v10.html
  *
  * Contributors:
  *    Hisashi MIYASHITA - initial API and implementation
  *******************************************************************************/

 package org.eclipse.actf.util.internal.httpproxy.core;

 import java.io.BufferedOutputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.Socket;

 import org.eclipse.actf.util.httpproxy.core.IClientConnection;
 import org.eclipse.actf.util.httpproxy.core.IHTTPHeader;
 import org.eclipse.actf.util.httpproxy.core.IHTTPRequestMessage;
 import org.eclipse.actf.util.httpproxy.core.IHTTPResponseMessage;
 import org.eclipse.actf.util.httpproxy.core.TimeoutException;
 import org.eclipse.actf.util.httpproxy.util.Logger;


 public abstract class ClientConnection implements Runnable, IClientConnection {
     private static final Logger LOGGER = Logger.getLogger(ClientConnection.class);

     private final ClientConnectionListener clientConnectionListener;

     private final int fQueueSize;

     // private final AsyncWorkManager fDispatchWorkMan;

     private long fKeepAlive;

     private Socket fClientSocket;

     private InputStream fClientIn;

     private BufferedOutputStream fClientOut;

     private HTTPRequestReader fReader;

     private RequestDispatcher fDispatcher;

     private long fLastReadTime = 0;

     private long fMessageSerial = 0;

     private boolean isHandlingRequest;

     private String connectionName;

     protected void setConnectionName(String name) {
         this.connectionName = name;
     }

     protected int getQueueSize() {
         return fQueueSize;
     }

     protected ClientConnection(ClientConnectionListener clientConnectionListener, int queueSize) {
         this.clientConnectionListener = clientConnectionListener;
         fQueueSize = queueSize;
         // fDispatchWorkMan = dispatchWorkMan;
     }

     /* (non-Javadoc)
 	 * @see org.eclipse.actf.util.httpproxy.core.IClientConnection#resetConnection()
 	 */
     public void resetConnection() {
         try {
             if (fClientSocket != null) {
                 if (fClientIn != null) {
                     fClientIn.close();
                 }
                 if (fClientOut != null) {
                     fClientOut.close();
                 }
                 if (!fClientSocket.isClosed()) {
                     fClientSocket.setSoLinger(true, 0);
                     fClientSocket.close();
                 }
                 fClientSocket = null;
             }
         } catch (IOException ex) {
             WARNING("Failed to shut down a client output connection (IOException): " + ex.getMessage());
             ex.printStackTrace();
         }
     }

     /* (non-Javadoc)
 	 * @see org.eclipse.actf.util.httpproxy.core.IClientConnection#close()
 	 */
     public String close() {
         if (LOGGER.isDebugEnabled()) {
             DEBUG("Shutdown a client socket: lastReadTime=" + fLastReadTime);
         }
         if (fDispatcher != null) {
             fDispatcher.close();
         }
         if (fClientSocket != null) {
             try {
                 fClientSocket.shutdownInput();
                 if (fClientIn != null) {
                     fClientIn.close();
                 }
             } catch (IOException ex) {
                 WARNING("Failed to close a client connection (IOException): " + ex.getMessage());
             }
         }

         String cNameTmp = connectionName;
         connectionName = null;
         fClientSocket = null;
         fClientOut = null;
         fReader = null;
         fDispatcher = null;
         // System.out.println("CLOSE");
         clientConnectionListener.connectionClosed(this);
         return cNameTmp;
     }

     protected void initInternal(Socket clientSock, long keepAlive, int timeout, RequestDispatcher dispatcher)
             throws IOException {
         fClientSocket = clientSock;
         fClientSocket.setSoTimeout(timeout);
         fClientIn = clientSock.getInputStream();
         fClientOut = new BufferedOutputStream(new SocketTimeoutRetryOutputStream(clientSock.getOutputStream()));
         fKeepAlive = keepAlive;
         fReader = dispatcher.createHTTPRequestReader(fClientIn);
         isHandlingRequest = false;
         fDispatcher = dispatcher;
         fDispatcher.start();
         DEBUG("Initialized");
     }

     protected HTTPRequestMessage createHTTPRequestMessage(long msgSerial) {
         return new HTTPRequestMessage(msgSerial);
     }

     /* (non-Javadoc)
 	 * @see org.eclipse.actf.util.httpproxy.core.IClientConnection#getClientSocket()
 	 */
     public Socket getClientSocket() {
         return fClientSocket;
     }

     /* (non-Javadoc)
 	 * @see org.eclipse.actf.util.httpproxy.core.IClientConnection#getCurrentServerGroupIndex()
 	 */
     public int getCurrentServerGroupIndex() {
         return clientConnectionListener.getCurrentServerGroupIndex();
     }

     /* (non-Javadoc)
 	 * @see org.eclipse.actf.util.httpproxy.core.IClientConnection#isHandlingRequest()
 	 */
     public synchronized boolean isHandlingRequest() {
         return isHandlingRequest;
     }

     private synchronized void setHandlingRequest(boolean b) {
         isHandlingRequest = b;
         if (!isHandlingRequest) {
             this.notify();
         }
     }

     private synchronized void waitHandlingRequestFinish() throws InterruptedException {
         while (isHandlingRequest) {
             this.wait();
         }
     }

     protected void requestReceived(IHTTPRequestMessage request) {
         // nop
     }

     /* (non-Javadoc)
 	 * @see org.eclipse.actf.util.httpproxy.core.IClientConnection#sendResponse(long, org.eclipse.actf.util.httpproxy.core.HTTPResponseMessage, boolean)
 	 */
     public void sendResponse(long timeout, IHTTPResponseMessage response, boolean readyToHandleRequest)
         throws InterruptedException, IOException, TimeoutException
     {
         if (fClientOut == null) return;

         if (!isHandlingRequest()) {
             System.err.println("*****INVALID STATE: response=" + response);
             System.exit(-1);
             return;
         }
         try {
             if (LOGGER.isDebugEnabled()) {
                 StringBuffer sb = new StringBuffer();
                 sb.append("Send a response to the client: serial=").append(response.getSerial());
                 sb.append(", tid=").append(response.getTid());
                 DEBUG(sb.toString());
             }

             // System.out.println(new String(response.getMessageBody()));
             // response.writeHeaders(System.out);
             if (LOGGER.isDebugEnabled()) {
                 StringBuffer sb = new StringBuffer();
                 sb.append("Response arrived: tid=").append(response.getTid());
                 sb.append(", msgSerial=").append(response.getSerial());
                 sb.append("\n_______________________________________________________\n");
                 try {
                     ByteArrayOutputStream ob = new ByteArrayOutputStream();
                     BifurcatedOutputStream o = new BifurcatedOutputStream(fClientOut, ob);
                     response.write(timeout, o);
                     ob.close();
                     sb.append(ob.toString());
                 } catch (Exception e) {
                     // nop
                 }
                 sb.append("\n~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~\n");
                 DEBUG(sb.toString());
             } else {
                 response.write(timeout, fClientOut);
             }
             fClientOut.flush();
             if (response.isConnectionToBeClosed()) {
                 fClientSocket.shutdownOutput();
                 throw new InterruptedException("This connection is no longer available.");
             }
 /*
         } catch (IOException e) {
             StringBuffer sb = new StringBuffer();
             sb.append("Failed to send a response to the client (IOException) : tid=");
             sb.append(response.getTid());
             sb.append(", message=").append(e.getMessage());
             LOGGER.warning(sb.toString());
 */
         } finally {
             assert isHandlingRequest();
             if (readyToHandleRequest) {
                 setHandlingRequest(false);
             }
         }
     }

     /* (non-Javadoc)
 	 * @see org.eclipse.actf.util.httpproxy.core.IClientConnection#sendResponse(long, org.eclipse.actf.util.httpproxy.core.IHTTPResponseMessage)
 	 */
     public void sendResponse(long timeout, IHTTPResponseMessage response)
         throws InterruptedException, TimeoutException, IOException
     {
         sendResponse(timeout, response, true);
     }

     /* (non-Javadoc)
 	 * @see org.eclipse.actf.util.httpproxy.core.IClientConnection#sendResponse(org.eclipse.actf.util.httpproxy.core.HTTPResponseInMemoryMessage)
 	 */
     public void sendResponse(IHTTPResponseMessage response)
         throws InterruptedException, IOException
     {
         try {
             sendResponse(0, response);
         } catch (TimeoutException e) {
             throw new IOException("Timeout: " + e.getMessage());
         }
     }

     private class TunnelThread extends Thread {
         private byte[] buffer;

         private static final int DEFUALT_TUNNEL_BUFFER_SIZE = 1024;

         private InputStream is;

         private OutputStream os;

         private boolean exited;

         synchronized private void exit() {
             exited = true;
             notifyAll();
         }

         synchronized public void waitExit() {
             while (true) {
                 try {
                     wait();
                     if (exited) {
                         return;
                     }
                 } catch (InterruptedException e) {
                 }
             }
         }

         public void run() {
             try {
                 while (true) {
                     try {
                         int size = is.available();
                         if (size < 0) {
                             break;
                         }
                         if (size > buffer.length) {
                             buffer = new byte[size];
                         }
                         size = is.read(buffer);
                         if (size < 0) {
                             break;
                         }
                         if (LOGGER.isDebugEnabled()) {
                             DEBUG("----TUNNEL--->\n"
                                   + ((size > 0) ? new String(buffer, 0, size) : "")
                                   +"\n----TUNNEL--->\n");
                         }
                         os.write(buffer, 0, size);
                         os.flush();
                     } catch (IOException e) {
                         break;
                     }
                 }
             } finally {
                 try {
                     is.close();
                     os.close();
                 } catch (IOException e) {
                 }
                 exit();
             }
         }

         TunnelThread(InputStream is, OutputStream os) {
             this.is = is;
             this.os = os;
             this.buffer = new byte[DEFUALT_TUNNEL_BUFFER_SIZE];
             this.exited = false;
         }

     }

     /* (non-Javadoc)
 	 * @see org.eclipse.actf.util.httpproxy.core.IClientConnection#allowTunnel(org.eclipse.actf.util.httpproxy.core.HTTPRequestMessage, org.eclipse.actf.util.httpproxy.core.ServerConnection, long)
 	 */
     public void allowTunnel(IHTTPRequestMessage req, ServerConnection sc, long timeout) throws InterruptedException, TimeoutException, IOException {
         if (LOGGER.isDebugEnabled()) {
             DEBUG("CONNECT to " + sc.toString());
         }
         try {
             sendResponse(timeout,
                          new HTTPResponseInMemoryMessage(req.getSerial(),
                                                          IHTTPHeader.HTTP_VERSION_1_0_A,
                                                          "200".getBytes(),
                                                          "OK".getBytes(),
                                                          IHTTPResponseMessage.EMPTY_BODY),
                          false);
             TunnelThread c2s = new TunnelThread(fClientIn, sc.getOutputStream());
             TunnelThread s2c = new TunnelThread(sc.getInputStream(), fClientOut);
             DEBUG("Tunnel is started.");
             c2s.start();
             s2c.start();
             c2s.waitExit();
             s2c.waitExit();
             DEBUG("Tunnel is finished.");
         } finally {
             setHandlingRequest(false);
         }
     }

     /* (non-Javadoc)
 	 * @see org.eclipse.actf.util.httpproxy.core.IClientConnection#rejectTunnel(org.eclipse.actf.util.httpproxy.core.HTTPRequestMessage, long)
 	 */
     public void rejectTunnel(IHTTPRequestMessage req, long timeout) throws InterruptedException, TimeoutException, IOException {
         sendResponse(timeout,
                 new HTTPResponseInMemoryMessage(req.getSerial(),
                         IHTTPHeader.HTTP_VERSION_1_0_A,
                         "405".getBytes(),
                         "Method Not Allowed".getBytes(),
                         IHTTPResponseMessage.EMPTY_BODY));
     }

     /* (non-Javadoc)
 	 * @see org.eclipse.actf.util.httpproxy.core.IClientConnection#run()
 	 */
     public void run() {
         DEBUG("Reader thread started");
         String cNameTmp;
         try {
             fLastReadTime = System.currentTimeMillis();

             setHandlingRequest(false);
             HTTPRequestMessage receivedRequest = null;
             long timeout = fKeepAlive;
             do {
                 waitHandlingRequestFinish();

                 // Read next request
                 if (receivedRequest == null) {
                     try {
                         setHandlingRequest(true);
                         long msgSerial = fMessageSerial + 1;
                         if (msgSerial == 0) {
                             fMessageSerial = 1;
                         }
                         receivedRequest = createHTTPRequestMessage(msgSerial);
                         if (timeout > 0) {
                             DEBUG("read w/ timeout: " + timeout);
                             fReader.readMessage(receivedRequest, timeout, false);
                         } else {
                             DEBUG("read w/o timeout");
                             fReader.readMessage(receivedRequest, false);
                         }
                         fLastReadTime = System.currentTimeMillis();
                         timeout = fKeepAlive;
                         requestReceived(receivedRequest);
                         // receivedRequest.setTid(fTidGen.getID());
                         fMessageSerial = msgSerial;
                         if (LOGGER.isDebugEnabled()) {
                             StringBuffer sb = new StringBuffer();
                             sb.append("Read a request: ").append(connectionName);
                             sb.append(", serial=").append(receivedRequest.getSerial());
                             sb.append(", tid=").append(receivedRequest.getTid());
                             DEBUG(sb.toString());
                         }
                         // receivedRequest.writeHeaders(System.out);
                     } catch (TimeoutException e) {
                         setHandlingRequest(false);
                         receivedRequest = null;
                         if (LOGGER.isDebugEnabled()) {
                             StringBuffer sb = new StringBuffer();
                             sb.append("Request reading timeout: lastReadTime=").append(fLastReadTime);
                             sb.append(", elapsed=").append(System.currentTimeMillis() - fLastReadTime);
                             DEBUG(sb.toString());
                         }
                     }
                 }

                 // Send a request to a server
                 if (receivedRequest != null) {
                     try {
                         fDispatcher.putRequest(receivedRequest, 10);
                         if (LOGGER.isDebugEnabled()) {
                             StringBuffer sb = new StringBuffer();
                             sb.append("Sent a request: serial=").append(receivedRequest.getSerial());
                             sb.append(", tid=").append(receivedRequest.getTid());
                             DEBUG(sb.toString());
                         }
                         receivedRequest = null;
                     } catch (InterruptedException e) {
                         // ignore, try again in the next loop
                         if (LOGGER.isDebugEnabled()) {
                             StringBuffer sb = new StringBuffer();
                             sb.append("Thread interrupted (Dispatcher.putRequest): serial=").append(
                                     receivedRequest.getSerial());
                             DEBUG(sb.toString());
                         }
                     } catch (TimeoutException e) {
                         // ignore, try again in the next loop
                         if (LOGGER.isDebugEnabled()) {
                             StringBuffer sb = new StringBuffer();
                             sb.append("Queue overflow (Dispatcher.putRequest): serial=").append(
                                     receivedRequest.getSerial());
                             DEBUG(sb.toString());
                         }
                     }
                 }
             } while (isHandlingRequest() || System.currentTimeMillis() - fLastReadTime < fKeepAlive);
             // WARNING("ZZZZ FINISHING: isHandlingRequest=" +
             // isHandlingRequest);
         } catch (InterruptedException interrupted) {
             // nop
             // WARNING("Interrupted");
         } catch (HTTPConnectionException e) {
         	DEBUG("The connection is closed by the peer.");
         } catch (IOException ioe) {
             // Exception from the InputStream
             if (LOGGER.isDebugEnabled()) {
                 DEBUG("Client connection is lost: " + ioe.getMessage());
                 ioe.printStackTrace();
             }
         } finally {
             cNameTmp = close();
         }
         DEBUG(cNameTmp, "Reader thread stopped");
     }

     private void DEBUG(String name, String msg) {
         if (LOGGER.isDebugEnabled()) {
             StringBuffer sb = new StringBuffer();
             sb.append(name).append(": ");
             sb.append(msg);
             LOGGER.debug(sb.toString());
         }
     }

     private final void DEBUG(String msg) {
         if (LOGGER.isDebugEnabled()) {
             StringBuffer sb = new StringBuffer();
             sb.append(connectionName).append(": ");
             sb.append(msg);
             LOGGER.debug(sb.toString());
         }
     }

     private final void WARNING(String msg) {
         StringBuffer sb = new StringBuffer();
         sb.append(connectionName).append(": ");
         sb.append(msg);
         LOGGER.warning(sb.toString());
     }
 }
	/*******************************************************************************
	* Copyright (c) 2007 IBM Corporation and Others
	* All rights reserved. This program and the accompanying materials
	* are made available under the terms of the Eclipse Public License v1.0
	* which accompanies this distribution, and is available at
	* http://www.eclipse.org/legal/epl-v10.html
	*
	* Contributors:
	* Hisashi MIYASHITA - initial API and implementation
	*******************************************************************************/

	package org.eclipse.actf.util.internal.httpproxy.core;

	import java.io.BufferedOutputStream;
	import java.io.ByteArrayOutputStream;
	import java.io.IOException;
	import java.io.InputStream;
	import java.io.OutputStream;
	import java.net.Socket;

	import org.eclipse.actf.util.httpproxy.core.IClientConnection;
	import org.eclipse.actf.util.httpproxy.core.IHTTPHeader;
	import org.eclipse.actf.util.httpproxy.core.IHTTPRequestMessage;
	import org.eclipse.actf.util.httpproxy.core.IHTTPResponseMessage;
	import org.eclipse.actf.util.httpproxy.core.TimeoutException;
	import org.eclipse.actf.util.httpproxy.util.Logger;




	public abstract class ClientConnection implements Runnable, IClientConnection {
	private static final Logger LOGGER = Logger.getLogger(ClientConnection.class);

	private final ClientConnectionListener clientConnectionListener;

	private final int fQueueSize;

	// private final AsyncWorkManager fDispatchWorkMan;

	private long fKeepAlive;

	private Socket fClientSocket;

	private InputStream fClientIn;

	private BufferedOutputStream fClientOut;

	private HTTPRequestReader fReader;

	private RequestDispatcher fDispatcher;

	private long fLastReadTime = 0;

	private long fMessageSerial = 0;

	private boolean isHandlingRequest;

	private String connectionName;

	protected void setConnectionName(String name) {
	this.connectionName = name;
	}

	protected int getQueueSize() {
	return fQueueSize;
	}

	protected ClientConnection(ClientConnectionListener clientConnectionListener, int queueSize) {
	this.clientConnectionListener = clientConnectionListener;
	fQueueSize = queueSize;
	// fDispatchWorkMan = dispatchWorkMan;
	}

	/* (non-Javadoc)
	* @see org.eclipse.actf.util.httpproxy.core.IClientConnection#resetConnection()
	*/
	public void resetConnection() {
	try {
	if (fClientSocket != null) {
	if (fClientIn != null) {
	fClientIn.close();
	}
	if (fClientOut != null) {
	fClientOut.close();
	}
	if (!fClientSocket.isClosed()) {
	fClientSocket.setSoLinger(true, 0);
	fClientSocket.close();
	}
	fClientSocket = null;
	}
	} catch (IOException ex) {
	WARNING("Failed to shut down a client output connection (IOException): " + ex.getMessage());
	ex.printStackTrace();
	}
	}

	/* (non-Javadoc)
	* @see org.eclipse.actf.util.httpproxy.core.IClientConnection#close()
	*/
	public String close() {
	if (LOGGER.isDebugEnabled()) {
	DEBUG("Shutdown a client socket: lastReadTime=" + fLastReadTime);
	}
	if (fDispatcher != null) {
	fDispatcher.close();
	}
	if (fClientSocket != null) {
	try {
	fClientSocket.shutdownInput();
	if (fClientIn != null) {
	fClientIn.close();
	}
	} catch (IOException ex) {
	WARNING("Failed to close a client connection (IOException): " + ex.getMessage());
	}
	}

	String cNameTmp = connectionName;
	connectionName = null;
	fClientSocket = null;
	fClientOut = null;
	fReader = null;
	fDispatcher = null;
	// System.out.println("CLOSE");
	clientConnectionListener.connectionClosed(this);
	return cNameTmp;
	}

	protected void initInternal(Socket clientSock, long keepAlive, int timeout, RequestDispatcher dispatcher)
	throws IOException {
	fClientSocket = clientSock;
	fClientSocket.setSoTimeout(timeout);
	fClientIn = clientSock.getInputStream();
	fClientOut = new BufferedOutputStream(new SocketTimeoutRetryOutputStream(clientSock.getOutputStream()));
	fKeepAlive = keepAlive;
	fReader = dispatcher.createHTTPRequestReader(fClientIn);
	isHandlingRequest = false;
	fDispatcher = dispatcher;
	fDispatcher.start();
	DEBUG("Initialized");
	}

	protected HTTPRequestMessage createHTTPRequestMessage(long msgSerial) {
	return new HTTPRequestMessage(msgSerial);
	}

	/* (non-Javadoc)
	* @see org.eclipse.actf.util.httpproxy.core.IClientConnection#getClientSocket()
	*/
	public Socket getClientSocket() {
	return fClientSocket;
	}

	/* (non-Javadoc)
	* @see org.eclipse.actf.util.httpproxy.core.IClientConnection#getCurrentServerGroupIndex()
	*/
	public int getCurrentServerGroupIndex() {
	return clientConnectionListener.getCurrentServerGroupIndex();
	}

	/* (non-Javadoc)
	* @see org.eclipse.actf.util.httpproxy.core.IClientConnection#isHandlingRequest()
	*/
	public synchronized boolean isHandlingRequest() {
	return isHandlingRequest;
	}

	private synchronized void setHandlingRequest(boolean b) {
	isHandlingRequest = b;
	if (!isHandlingRequest) {
	this.notify();
	}
	}

	private synchronized void waitHandlingRequestFinish() throws InterruptedException {
	while (isHandlingRequest) {
	this.wait();
	}
	}

	protected void requestReceived(IHTTPRequestMessage request) {
	// nop
	}

	/* (non-Javadoc)
	* @see org.eclipse.actf.util.httpproxy.core.IClientConnection#sendResponse(long, org.eclipse.actf.util.httpproxy.core.HTTPResponseMessage, boolean)
	*/
	public void sendResponse(long timeout, IHTTPResponseMessage response, boolean readyToHandleRequest)
	throws InterruptedException, IOException, TimeoutException
	{
	if (fClientOut == null) return;

	if (!isHandlingRequest()) {
	System.err.println("*****INVALID STATE: response=" + response);
	System.exit(-1);
	return;
	}
	try {
	if (LOGGER.isDebugEnabled()) {
	StringBuffer sb = new StringBuffer();
	sb.append("Send a response to the client: serial=").append(response.getSerial());
	sb.append(", tid=").append(response.getTid());
	DEBUG(sb.toString());
	}

	// System.out.println(new String(response.getMessageBody()));
	// response.writeHeaders(System.out);
	if (LOGGER.isDebugEnabled()) {
	StringBuffer sb = new StringBuffer();
	sb.append("Response arrived: tid=").append(response.getTid());
	sb.append(", msgSerial=").append(response.getSerial());
	sb.append("\n_______________________________________________________\n");
	try {
	ByteArrayOutputStream ob = new ByteArrayOutputStream();
	BifurcatedOutputStream o = new BifurcatedOutputStream(fClientOut, ob);
	response.write(timeout, o);
	ob.close();
	sb.append(ob.toString());
	} catch (Exception e) {
	// nop
	}
	sb.append("\n~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~\n");
	DEBUG(sb.toString());
	} else {
	response.write(timeout, fClientOut);
	}
	fClientOut.flush();
	if (response.isConnectionToBeClosed()) {
	fClientSocket.shutdownOutput();
	throw new InterruptedException("This connection is no longer available.");
	}
	/*
	} catch (IOException e) {
	StringBuffer sb = new StringBuffer();
	sb.append("Failed to send a response to the client (IOException) : tid=");
	sb.append(response.getTid());
	sb.append(", message=").append(e.getMessage());
	LOGGER.warning(sb.toString());
	*/
	} finally {
	assert isHandlingRequest();
	if (readyToHandleRequest) {
	setHandlingRequest(false);
	}
	}
	}

	/* (non-Javadoc)
	* @see org.eclipse.actf.util.httpproxy.core.IClientConnection#sendResponse(long, org.eclipse.actf.util.httpproxy.core.IHTTPResponseMessage)
	*/
	public void sendResponse(long timeout, IHTTPResponseMessage response)
	throws InterruptedException, TimeoutException, IOException
	{
	sendResponse(timeout, response, true);
	}

	/* (non-Javadoc)
	* @see org.eclipse.actf.util.httpproxy.core.IClientConnection#sendResponse(org.eclipse.actf.util.httpproxy.core.HTTPResponseInMemoryMessage)
	*/
	public void sendResponse(IHTTPResponseMessage response)
	throws InterruptedException, IOException
	{
	try {
	sendResponse(0, response);
	} catch (TimeoutException e) {
	throw new IOException("Timeout: " + e.getMessage());
	}
	}

	private class TunnelThread extends Thread {
	private byte[] buffer;

	private static final int DEFUALT_TUNNEL_BUFFER_SIZE = 1024;

	private InputStream is;

	private OutputStream os;

	private boolean exited;

	synchronized private void exit() {
	exited = true;
	notifyAll();
	}

	synchronized public void waitExit() {
	while (true) {
	try {
	wait();
	if (exited) {
	return;
	}
	} catch (InterruptedException e) {
	}
	}
	}

	public void run() {
	try {
	while (true) {
	try {
	int size = is.available();
	if (size < 0) {
	break;
	}
	if (size > buffer.length) {
	buffer = new byte[size];
	}
	size = is.read(buffer);
	if (size < 0) {
	break;
	}
	if (LOGGER.isDebugEnabled()) {
	DEBUG("----TUNNEL--->\n"
	+ ((size > 0) ? new String(buffer, 0, size) : "")
	+"\n----TUNNEL--->\n");
	}
	os.write(buffer, 0, size);
	os.flush();
	} catch (IOException e) {
	break;
	}
	}
	} finally {
	try {
	is.close();
	os.close();
	} catch (IOException e) {
	}
	exit();
	}
	}

	TunnelThread(InputStream is, OutputStream os) {
	this.is = is;
	this.os = os;
	this.buffer = new byte[DEFUALT_TUNNEL_BUFFER_SIZE];
	this.exited = false;
	}

	}

	/* (non-Javadoc)
	* @see org.eclipse.actf.util.httpproxy.core.IClientConnection#allowTunnel(org.eclipse.actf.util.httpproxy.core.HTTPRequestMessage, org.eclipse.actf.util.httpproxy.core.ServerConnection, long)
	*/
	public void allowTunnel(IHTTPRequestMessage req, ServerConnection sc, long timeout) throws InterruptedException, TimeoutException, IOException {
	if (LOGGER.isDebugEnabled()) {
	DEBUG("CONNECT to " + sc.toString());
	}
	try {
	sendResponse(timeout,
	new HTTPResponseInMemoryMessage(req.getSerial(),
	IHTTPHeader.HTTP_VERSION_1_0_A,
	"200".getBytes(),
	"OK".getBytes(),
	IHTTPResponseMessage.EMPTY_BODY),
	false);
	TunnelThread c2s = new TunnelThread(fClientIn, sc.getOutputStream());
	TunnelThread s2c = new TunnelThread(sc.getInputStream(), fClientOut);
	DEBUG("Tunnel is started.");
	c2s.start();
	s2c.start();
	c2s.waitExit();
	s2c.waitExit();
	DEBUG("Tunnel is finished.");
	} finally {
	setHandlingRequest(false);
	}
	}

	/* (non-Javadoc)
	* @see org.eclipse.actf.util.httpproxy.core.IClientConnection#rejectTunnel(org.eclipse.actf.util.httpproxy.core.HTTPRequestMessage, long)
	*/
	public void rejectTunnel(IHTTPRequestMessage req, long timeout) throws InterruptedException, TimeoutException, IOException {
	sendResponse(timeout,
	new HTTPResponseInMemoryMessage(req.getSerial(),
	IHTTPHeader.HTTP_VERSION_1_0_A,
	"405".getBytes(),
	"Method Not Allowed".getBytes(),
	IHTTPResponseMessage.EMPTY_BODY));
	}

	/* (non-Javadoc)
	* @see org.eclipse.actf.util.httpproxy.core.IClientConnection#run()
	*/
	public void run() {
	DEBUG("Reader thread started");
	String cNameTmp;
	try {
	fLastReadTime = System.currentTimeMillis();

	setHandlingRequest(false);
	HTTPRequestMessage receivedRequest = null;
	long timeout = fKeepAlive;
	do {
	waitHandlingRequestFinish();

	// Read next request
	if (receivedRequest == null) {
	try {
	setHandlingRequest(true);
	long msgSerial = fMessageSerial + 1;
	if (msgSerial == 0) {
	fMessageSerial = 1;
	}
	receivedRequest = createHTTPRequestMessage(msgSerial);
	if (timeout > 0) {
	DEBUG("read w/ timeout: " + timeout);
	fReader.readMessage(receivedRequest, timeout, false);
	} else {
	DEBUG("read w/o timeout");
	fReader.readMessage(receivedRequest, false);
	}
	fLastReadTime = System.currentTimeMillis();
	timeout = fKeepAlive;
	requestReceived(receivedRequest);
	// receivedRequest.setTid(fTidGen.getID());
	fMessageSerial = msgSerial;
	if (LOGGER.isDebugEnabled()) {
	StringBuffer sb = new StringBuffer();
	sb.append("Read a request: ").append(connectionName);
	sb.append(", serial=").append(receivedRequest.getSerial());
	sb.append(", tid=").append(receivedRequest.getTid());
	DEBUG(sb.toString());
	}
	// receivedRequest.writeHeaders(System.out);
	} catch (TimeoutException e) {
	setHandlingRequest(false);
	receivedRequest = null;
	if (LOGGER.isDebugEnabled()) {
	StringBuffer sb = new StringBuffer();
	sb.append("Request reading timeout: lastReadTime=").append(fLastReadTime);
	sb.append(", elapsed=").append(System.currentTimeMillis() - fLastReadTime);
	DEBUG(sb.toString());
	}
	}
	}

	// Send a request to a server
	if (receivedRequest != null) {
	try {
	fDispatcher.putRequest(receivedRequest, 10);
	if (LOGGER.isDebugEnabled()) {
	StringBuffer sb = new StringBuffer();
	sb.append("Sent a request: serial=").append(receivedRequest.getSerial());
	sb.append(", tid=").append(receivedRequest.getTid());
	DEBUG(sb.toString());
	}
	receivedRequest = null;
	} catch (InterruptedException e) {
	// ignore, try again in the next loop
	if (LOGGER.isDebugEnabled()) {
	StringBuffer sb = new StringBuffer();
	sb.append("Thread interrupted (Dispatcher.putRequest): serial=").append(
	receivedRequest.getSerial());
	DEBUG(sb.toString());
	}
	} catch (TimeoutException e) {
	// ignore, try again in the next loop
	if (LOGGER.isDebugEnabled()) {
	StringBuffer sb = new StringBuffer();
	sb.append("Queue overflow (Dispatcher.putRequest): serial=").append(
	receivedRequest.getSerial());
	DEBUG(sb.toString());
	}
	}
	}
	} while (isHandlingRequest() \|\| System.currentTimeMillis() - fLastReadTime < fKeepAlive);
	// WARNING("ZZZZ FINISHING: isHandlingRequest=" +
	// isHandlingRequest);
	} catch (InterruptedException interrupted) {
	// nop
	// WARNING("Interrupted");
	} catch (HTTPConnectionException e) {
	DEBUG("The connection is closed by the peer.");
	} catch (IOException ioe) {
	// Exception from the InputStream
	if (LOGGER.isDebugEnabled()) {
	DEBUG("Client connection is lost: " + ioe.getMessage());
	ioe.printStackTrace();
	}
	} finally {
	cNameTmp = close();
	}
	DEBUG(cNameTmp, "Reader thread stopped");
	}

	private void DEBUG(String name, String msg) {
	if (LOGGER.isDebugEnabled()) {
	StringBuffer sb = new StringBuffer();
	sb.append(name).append(": ");
	sb.append(msg);
	LOGGER.debug(sb.toString());
	}
	}

	private final void DEBUG(String msg) {
	if (LOGGER.isDebugEnabled()) {
	StringBuffer sb = new StringBuffer();
	sb.append(connectionName).append(": ");
	sb.append(msg);
	LOGGER.debug(sb.toString());
	}
	}

	private final void WARNING(String msg) {
	StringBuffer sb = new StringBuffer();
	sb.append(connectionName).append(": ");
	sb.append(msg);
	LOGGER.warning(sb.toString());
	}
	}