jetty-rhttp/jetty-rhttp-gateway/src/main/java/org/eclipse/jetty/rhttp/gateway/ConnectorServlet.java - jetty - Git at Google

 //
 //  ========================================================================
 //  Copyright (c) 1995-2017 Mort Bay Consulting Pty. Ltd.
 //  ------------------------------------------------------------------------
 //  All rights reserved. This program and the accompanying materials
 //  are made available under the terms of the Eclipse Public License v1.0
 //  and Apache License v2.0 which accompanies this distribution.
 //
 //      The Eclipse Public License is available at
 //      http://www.eclipse.org/legal/epl-v10.html
 //
 //      The Apache License v2.0 is available at
 //      http://www.opensource.org/licenses/apache2.0.php
 //
 //  You may elect to redistribute this code under either of these licenses.
 //  ========================================================================
 //

 package org.eclipse.jetty.rhttp.gateway;

 import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;

 import javax.servlet.ServletException;
 import javax.servlet.ServletOutputStream;
 import javax.servlet.http.HttpServlet;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;

 import org.eclipse.jetty.rhttp.client.RHTTPRequest;
 import org.eclipse.jetty.rhttp.client.RHTTPResponse;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;

 /**
  * The servlet that handles the communication with the gateway clients.
  * @version $Revision$ $Date$
  */
 public class ConnectorServlet extends HttpServlet
 {
     private final Logger logger = Log.getLogger(getClass().toString());
     private final TargetIdRetriever targetIdRetriever = new StandardTargetIdRetriever();
     private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
     private final ConcurrentMap<String, Future<?>> expirations = new ConcurrentHashMap<String, Future<?>>();
     private final Gateway gateway;
     private long clientTimeout=15000;

     public ConnectorServlet(Gateway gateway)
     {
         this.gateway = gateway;
     }

     @Override
     public void init() throws ServletException
     {
         String t = getInitParameter("clientTimeout");
         if (t!=null && !"".equals(t))
             clientTimeout=Long.parseLong(t);
     }

     @Override
     protected void service(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException
     {
         String targetId = targetIdRetriever.retrieveTargetId(request);

         String uri = request.getRequestURI();
         String path = uri.substring(request.getServletPath().length());
         String[] segments = path.split("/");
         if (segments.length < 3)
             throw new ServletException("Invalid request to " + getClass().getSimpleName() + ": " + uri);

         String action = segments[2];
         if ("handshake".equals(action))
             serviceHandshake(targetId, request, response);
         else if ("connect".equals(action))
             serviceConnect(targetId, request, response);
         else if ("deliver".equals(action))
             serviceDeliver(targetId, request, response);
         else if ("disconnect".equals(action))
             serviceDisconnect(targetId, request, response);
         else
             throw new ServletException("Invalid request to " + getClass().getSimpleName() + ": " + uri);
     }

     private void serviceHandshake(String targetId, HttpServletRequest httpRequest, HttpServletResponse httpResponse) throws IOException
     {
         ClientDelegate client = gateway.getClientDelegate(targetId);
         if (client != null)
             throw new IOException("Client with targetId " + targetId + " is already connected");

         client = gateway.newClientDelegate(targetId);
         ClientDelegate existing = gateway.addClientDelegate(targetId, client);
         if (existing != null)
             throw new IOException("Client with targetId " + targetId + " is already connected");

         flush(client, httpRequest, httpResponse);
     }

     private void flush(ClientDelegate client, HttpServletRequest httpRequest, HttpServletResponse httpResponse) throws IOException
     {
         List<RHTTPRequest> requests = client.process(httpRequest);
         if (requests != null)
         {
             // Schedule before sending the requests, to avoid that the remote client
             // reconnects before we have scheduled the expiration timeout.
             if (!client.isClosed())
                 schedule(client);

             ServletOutputStream output = httpResponse.getOutputStream();
             for (RHTTPRequest request : requests)
                 output.write(request.getFrameBytes());
             // I could count the framed bytes of all requests and set a Content-Length header,
             // but the implementation of ServletOutputStream takes care of everything:
             // if the request was HTTP/1.1, then flushing result in a chunked response, but the
             // client know how to handle it; if the request was HTTP/1.0, then no chunking.
             // To avoid chunking in HTTP/1.1 I must set the Content-Length header.
             output.flush();
             logger.debug("Delivered to device {} requests {} ", client.getTargetId(), requests);
         }
     }

     private void schedule(ClientDelegate client)
     {
         Future<?> task = scheduler.schedule(new ClientExpirationTask(client), clientTimeout, TimeUnit.MILLISECONDS);
         Future<?> existing = expirations.put(client.getTargetId(), task);
         assert existing == null;
     }

     private void unschedule(String targetId)
     {
         Future<?> task = expirations.remove(targetId);
         if (task != null)
             task.cancel(false);
     }

     private void serviceConnect(String targetId, HttpServletRequest httpRequest, HttpServletResponse httpResponse) throws IOException
     {
         unschedule(targetId);

         ClientDelegate client = gateway.getClientDelegate(targetId);
         if (client == null)
         {
             // Expired client tries to connect without handshake
             httpResponse.sendError(HttpServletResponse.SC_UNAUTHORIZED);
             return;
         }

         flush(client, httpRequest, httpResponse);

         if (client.isClosed())
             gateway.removeClientDelegate(targetId);
     }

     private void expireConnect(ClientDelegate client, long time)
     {
         String targetId = client.getTargetId();
         logger.info("Client with targetId {} missing, last seen {} ms ago, closing it", targetId, System.currentTimeMillis() - time);
         client.close();
         // If the client expired, means that it did not connect,
         // so there no request to resume, and we cleanup here
         // (while normally this cleanup is done in serviceConnect())
         unschedule(targetId);
         gateway.removeClientDelegate(targetId);
     }

     private void serviceDeliver(String targetId, HttpServletRequest httpRequest, HttpServletResponse httpResponse) throws ServletException, IOException
     {
         if (gateway.getClientDelegate(targetId) == null)
         {
             // Expired client tries to deliver without handshake
             httpResponse.sendError(HttpServletResponse.SC_UNAUTHORIZED);
             return;
         }

         byte[] body = Utils.read(httpRequest.getInputStream());

         RHTTPResponse response = RHTTPResponse.fromFrameBytes(body);

         ExternalRequest externalRequest = gateway.removeExternalRequest(response.getId());
         if (externalRequest != null)
         {
             externalRequest.respond(response);
             logger.debug("Deliver request from device {}, gateway request {}, response {}", new Object[] {targetId, externalRequest, response});
         }
         else
         {
             // We can arrive here for a race with the continuation expiration, which expired just before
             // the gateway client responded with a valid response; log this case ignore it.
             logger.debug("Deliver request from device {}, missing gateway request, response {}", targetId, response);
         }
     }

     private void serviceDisconnect(String targetId, HttpServletRequest request, HttpServletResponse response)
     {
         // Do not remove the ClientDelegate from the gateway here,
         // since closing the ClientDelegate will resume the connect request
         // and we remove the ClientDelegate from the gateway there
         ClientDelegate client = gateway.getClientDelegate(targetId);
         if (client != null)
             client.close();
     }

     private class ClientExpirationTask implements Runnable
     {
         private final long time = System.currentTimeMillis();
         private final ClientDelegate client;

         public ClientExpirationTask(ClientDelegate client)
         {
             this.client = client;
         }

         public void run()
         {
             expireConnect(client, time);
         }
     }
 }
	//
	// ========================================================================
	// Copyright (c) 1995-2017 Mort Bay Consulting Pty. Ltd.
	// ------------------------------------------------------------------------
	// All rights reserved. This program and the accompanying materials
	// are made available under the terms of the Eclipse Public License v1.0
	// and Apache License v2.0 which accompanies this distribution.
	//
	// The Eclipse Public License is available at
	// http://www.eclipse.org/legal/epl-v10.html
	//
	// The Apache License v2.0 is available at
	// http://www.opensource.org/licenses/apache2.0.php
	//
	// You may elect to redistribute this code under either of these licenses.
	// ========================================================================
	//

	package org.eclipse.jetty.rhttp.gateway;

	import java.io.IOException;
	import java.util.List;
	import java.util.concurrent.ConcurrentHashMap;
	import java.util.concurrent.ConcurrentMap;
	import java.util.concurrent.Executors;
	import java.util.concurrent.Future;
	import java.util.concurrent.ScheduledExecutorService;
	import java.util.concurrent.TimeUnit;

	import javax.servlet.ServletException;
	import javax.servlet.ServletOutputStream;
	import javax.servlet.http.HttpServlet;
	import javax.servlet.http.HttpServletRequest;
	import javax.servlet.http.HttpServletResponse;

	import org.eclipse.jetty.rhttp.client.RHTTPRequest;
	import org.eclipse.jetty.rhttp.client.RHTTPResponse;
	import org.eclipse.jetty.util.log.Log;
	import org.eclipse.jetty.util.log.Logger;

	/**
	* The servlet that handles the communication with the gateway clients.
	* @version $Revision$ $Date$
	*/
	public class ConnectorServlet extends HttpServlet
	{
	private final Logger logger = Log.getLogger(getClass().toString());
	private final TargetIdRetriever targetIdRetriever = new StandardTargetIdRetriever();
	private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
	private final ConcurrentMap<String, Future<?>> expirations = new ConcurrentHashMap<String, Future<?>>();
	private final Gateway gateway;
	private long clientTimeout=15000;

	public ConnectorServlet(Gateway gateway)
	{
	this.gateway = gateway;
	}

	@Override
	public void init() throws ServletException
	{
	String t = getInitParameter("clientTimeout");
	if (t!=null && !"".equals(t))
	clientTimeout=Long.parseLong(t);
	}

	@Override
	protected void service(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException
	{
	String targetId = targetIdRetriever.retrieveTargetId(request);

	String uri = request.getRequestURI();
	String path = uri.substring(request.getServletPath().length());
	String[] segments = path.split("/");
	if (segments.length < 3)
	throw new ServletException("Invalid request to " + getClass().getSimpleName() + ": " + uri);

	String action = segments[2];
	if ("handshake".equals(action))
	serviceHandshake(targetId, request, response);
	else if ("connect".equals(action))
	serviceConnect(targetId, request, response);
	else if ("deliver".equals(action))
	serviceDeliver(targetId, request, response);
	else if ("disconnect".equals(action))
	serviceDisconnect(targetId, request, response);
	else
	throw new ServletException("Invalid request to " + getClass().getSimpleName() + ": " + uri);
	}

	private void serviceHandshake(String targetId, HttpServletRequest httpRequest, HttpServletResponse httpResponse) throws IOException
	{
	ClientDelegate client = gateway.getClientDelegate(targetId);
	if (client != null)
	throw new IOException("Client with targetId " + targetId + " is already connected");

	client = gateway.newClientDelegate(targetId);
	ClientDelegate existing = gateway.addClientDelegate(targetId, client);
	if (existing != null)
	throw new IOException("Client with targetId " + targetId + " is already connected");

	flush(client, httpRequest, httpResponse);
	}

	private void flush(ClientDelegate client, HttpServletRequest httpRequest, HttpServletResponse httpResponse) throws IOException
	{
	List<RHTTPRequest> requests = client.process(httpRequest);
	if (requests != null)
	{
	// Schedule before sending the requests, to avoid that the remote client
	// reconnects before we have scheduled the expiration timeout.
	if (!client.isClosed())
	schedule(client);

	ServletOutputStream output = httpResponse.getOutputStream();
	for (RHTTPRequest request : requests)
	output.write(request.getFrameBytes());
	// I could count the framed bytes of all requests and set a Content-Length header,
	// but the implementation of ServletOutputStream takes care of everything:
	// if the request was HTTP/1.1, then flushing result in a chunked response, but the
	// client know how to handle it; if the request was HTTP/1.0, then no chunking.
	// To avoid chunking in HTTP/1.1 I must set the Content-Length header.
	output.flush();
	logger.debug("Delivered to device {} requests {} ", client.getTargetId(), requests);
	}
	}

	private void schedule(ClientDelegate client)
	{
	Future<?> task = scheduler.schedule(new ClientExpirationTask(client), clientTimeout, TimeUnit.MILLISECONDS);
	Future<?> existing = expirations.put(client.getTargetId(), task);
	assert existing == null;
	}

	private void unschedule(String targetId)
	{
	Future<?> task = expirations.remove(targetId);
	if (task != null)
	task.cancel(false);
	}

	private void serviceConnect(String targetId, HttpServletRequest httpRequest, HttpServletResponse httpResponse) throws IOException
	{
	unschedule(targetId);

	ClientDelegate client = gateway.getClientDelegate(targetId);
	if (client == null)
	{
	// Expired client tries to connect without handshake
	httpResponse.sendError(HttpServletResponse.SC_UNAUTHORIZED);
	return;
	}

	flush(client, httpRequest, httpResponse);

	if (client.isClosed())
	gateway.removeClientDelegate(targetId);
	}

	private void expireConnect(ClientDelegate client, long time)
	{
	String targetId = client.getTargetId();
	logger.info("Client with targetId {} missing, last seen {} ms ago, closing it", targetId, System.currentTimeMillis() - time);
	client.close();
	// If the client expired, means that it did not connect,
	// so there no request to resume, and we cleanup here
	// (while normally this cleanup is done in serviceConnect())
	unschedule(targetId);
	gateway.removeClientDelegate(targetId);
	}

	private void serviceDeliver(String targetId, HttpServletRequest httpRequest, HttpServletResponse httpResponse) throws ServletException, IOException
	{
	if (gateway.getClientDelegate(targetId) == null)
	{
	// Expired client tries to deliver without handshake
	httpResponse.sendError(HttpServletResponse.SC_UNAUTHORIZED);
	return;
	}

	byte[] body = Utils.read(httpRequest.getInputStream());

	RHTTPResponse response = RHTTPResponse.fromFrameBytes(body);

	ExternalRequest externalRequest = gateway.removeExternalRequest(response.getId());
	if (externalRequest != null)
	{
	externalRequest.respond(response);
	logger.debug("Deliver request from device {}, gateway request {}, response {}", new Object[] {targetId, externalRequest, response});
	}
	else
	{
	// We can arrive here for a race with the continuation expiration, which expired just before
	// the gateway client responded with a valid response; log this case ignore it.
	logger.debug("Deliver request from device {}, missing gateway request, response {}", targetId, response);
	}
	}

	private void serviceDisconnect(String targetId, HttpServletRequest request, HttpServletResponse response)
	{
	// Do not remove the ClientDelegate from the gateway here,
	// since closing the ClientDelegate will resume the connect request
	// and we remove the ClientDelegate from the gateway there
	ClientDelegate client = gateway.getClientDelegate(targetId);
	if (client != null)
	client.close();
	}

	private class ClientExpirationTask implements Runnable
	{
	private final long time = System.currentTimeMillis();
	private final ClientDelegate client;

	public ClientExpirationTask(ClientDelegate client)
	{
	this.client = client;
	}

	public void run()
	{
	expireConnect(client, time);
	}
	}
	}