//
   //  ========================================================================
   //  Copyright (c) 1995-2012 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.spdy.server.http;
  
  import java.nio.ByteBuffer;
  import java.util.LinkedList;
  import java.util.Queue;
  
  import org.eclipse.jetty.http.HttpHeader;
  import org.eclipse.jetty.http.HttpMethod;
  import org.eclipse.jetty.http.HttpVersion;
  import org.eclipse.jetty.io.EndPoint;
  import org.eclipse.jetty.server.Connector;
  import org.eclipse.jetty.server.HttpChannel;
  import org.eclipse.jetty.server.HttpChannelConfig;
  import org.eclipse.jetty.server.HttpTransport;
  import org.eclipse.jetty.spdy.api.ByteBufferDataInfo;
  import org.eclipse.jetty.spdy.api.DataInfo;
  import org.eclipse.jetty.spdy.api.Stream;
  import org.eclipse.jetty.util.Fields;
  import org.eclipse.jetty.util.log.Log;
  import org.eclipse.jetty.util.log.Logger;
  
  public class HttpChannelOverSPDY extends HttpChannel<DataInfo>
  {
      private static final Logger LOG = Log.getLogger(HttpChannelOverSPDY.class);
  
      private final Queue<Runnable> tasks = new LinkedList<>();
      private final Stream stream;
      private boolean dispatched; // Guarded by synchronization on tasks
      private boolean headersComplete;
  
      public HttpChannelOverSPDY(Connector connector, HttpChannelConfig configuration, EndPoint endPoint, HttpTransport transport, HttpInputOverSPDY input, Stream stream)
      {
          super(connector, configuration, endPoint, transport, input);
          this.stream = stream;
      }
  
      @Override
      public boolean headerComplete()
      {
          headersComplete = true;
          return super.headerComplete();
      }
  
      private void post(Runnable task)
      {
          synchronized (tasks)
          {
              LOG.debug("Posting task {}", task);
              tasks.offer(task);
              dispatch();
          }
      }
  
      private void dispatch()
      {
          synchronized (tasks)
          {
              if (dispatched)
                  return;
  
              final Runnable task = tasks.poll();
              if (task != null)
              {
                  dispatched = true;
                  LOG.debug("Dispatching task {}", task);
                  execute(new Runnable()
                  {
                      @Override
                      public void run()
                      {
                          LOG.debug("Executing task {}", task);
                          task.run();
                          LOG.debug("Completing task {}", task);
                          dispatched = false;
                          dispatch();
                      }
                  });
              }
          }
      }
  
     public void requestStart(final Fields headers, final boolean endRequest)
     {
         if (!headers.isEmpty())
             requestHeaders(headers, endRequest);
     }
 
     public void requestHeaders(Fields headers, boolean endRequest)
     {
         boolean proceed = performBeginRequest(headers);
         if (!proceed)
             return;
 
         performHeaders(headers);
 
         if (endRequest)
         {
             if (headerComplete())
                 post(this);
             if (messageComplete(-1))
                 post(this);
         }
     }
 
     public void requestContent(final DataInfo dataInfo, boolean endRequest)
     {
         if (!headersComplete)
         {
             if (headerComplete())
                 post(this);
         }
 
         LOG.debug("HTTP > {} bytes of content", dataInfo.length());
 
         // We need to copy the dataInfo since we do not know when its bytes
         // will be consumed. When the copy is consumed, we consume also the
         // original, so the implementation can send a window update.
         ByteBuffer copyByteBuffer = dataInfo.asByteBuffer(false);
         ByteBufferDataInfo copyDataInfo = new ByteBufferDataInfo(copyByteBuffer, dataInfo.isClose(), dataInfo.isCompress())
         {
             @Override
             public void consume(int delta)
             {
                 super.consume(delta);
                 dataInfo.consume(delta);
             }
         };
         LOG.debug("Queuing last={} content {}", endRequest, copyDataInfo);
 
         if (content(copyDataInfo))
             post(this);
 
         if (endRequest)
         {
             if (messageComplete(-1))
                 post(this);
         }
     }
 
     private boolean performBeginRequest(Fields headers)
     {
         short version = stream.getSession().getVersion();
         Fields.Field methodHeader = headers.get(HTTPSPDYHeader.METHOD.name(version));
         Fields.Field uriHeader = headers.get(HTTPSPDYHeader.URI.name(version));
         Fields.Field versionHeader = headers.get(HTTPSPDYHeader.VERSION.name(version));
 
         if (methodHeader == null || uriHeader == null || versionHeader == null)
         {
             badMessage(400, "Missing required request line elements");
             return false;
         }
 
         HttpMethod httpMethod = HttpMethod.fromString(methodHeader.value());
         HttpVersion httpVersion = HttpVersion.fromString(versionHeader.value());
         String uriString = uriHeader.value();
 
         LOG.debug("HTTP > {} {} {}", httpMethod, uriString, httpVersion);
         startRequest(httpMethod, httpMethod.asString(), uriString, httpVersion);
 
         Fields.Field schemeHeader = headers.get(HTTPSPDYHeader.SCHEME.name(version));
         if (schemeHeader != null)
             getRequest().setScheme(schemeHeader.value());
         return true;
     }
 
     private void performHeaders(Fields headers)
     {
         for (Fields.Field header : headers)
         {
             String name = header.name();
             HttpHeader httpHeader = HttpHeader.CACHE.get(name);
 
             // Skip special SPDY headers, unless it's the "host" header
             HTTPSPDYHeader specialHeader = HTTPSPDYHeader.from(stream.getSession().getVersion(), name);
             if (specialHeader != null)
             {
                 if (specialHeader == HTTPSPDYHeader.HOST)
                     name = "host";
                 else
                     continue;
             }
 
             switch (name)
             {
                 case "connection":
                 case "keep-alive":
                 case "proxy-connection":
                 case "transfer-encoding":
                 {
                     // Spec says to ignore these headers
                     continue;
                 }
                 default:
                 {
                     // Spec says headers must be single valued
                     String value = header.value();
                     LOG.debug("HTTP > {}: {}", name, value);
                     parsedHeader(httpHeader, name, value);
                     break;
                 }
             }
         }
     }
 }