content: /*
   * The Apache Software License, Version 1.1
   *
   * Copyright (c) 2001-2004 Caucho Technology, Inc.  All rights reserved.
   *
   * Redistribution and use in source and binary forms, with or without
   * modification, are permitted provided that the following conditions
   * are met:
   *
   * 1. Redistributions of source code must retain the above copyright
   *    notice, this list of conditions and the following disclaimer.
   *
   * 2. Redistributions in binary form must reproduce the above copyright
   *    notice, this list of conditions and the following disclaimer in
   *    the documentation and/or other materials provided with the
   *    distribution.
   *
   * 3. The end-user documentation included with the redistribution, if
   *    any, must include the following acknowlegement:
   *       "This product includes software developed by the
   *        Caucho Technology (http://www.caucho.com/)."
   *    Alternately, this acknowlegement may appear in the software itself,
   *    if and wherever such third-party acknowlegements normally appear.
   *
   * 4. The names "Hessian", "Resin", and "Caucho" must not be used to
   *    endorse or promote products derived from this software without prior
   *    written permission. For written permission, please contact
   *    info@caucho.com.
   *
   * 5. Products derived from this software may not be called "Resin"
   *    nor may "Resin" appear in their names without prior written
   *    permission of Caucho Technology.
   *
   * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
   * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
   * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
   * DISCLAIMED.  IN NO EVENT SHALL CAUCHO TECHNOLOGY OR ITS CONTRIBUTORS
   * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY,
   * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
   * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
   * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
   * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
   * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN
   * IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
   *
   * @author Scott Ferguson
   */
  
  package com.caucho.hessian.mux;
  
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.OutputStream;
  
  /**
   * Hessian Mux, a peer-to-peer protocol.
   */
  public class MuxServer {
    private Object READ_LOCK = new Object();
    private Object WRITE_LOCK = new Object();
    
    private InputStream is;
    private OutputStream os;
    private boolean isClient;
  
    private transient boolean isClosed;
    
    // channels that have data ready.
    private boolean inputReady[] = new boolean[4];
  
    // true if there's a thread already reading
    private boolean isReadLocked;
    // true if there's a thread already writing
    private boolean isWriteLocked;
  
    /**
     * Null argument constructor.
     */
    public MuxServer()
    {
    }
  
    /**
     * Create a new multiplexor with input and output streams.
     *
     * @param is the underlying input stream
     * @param os the underlying output stream
     * @param isClient true if this is the connection client.
     */
    public MuxServer(InputStream is, OutputStream os, boolean isClient)
    {
      init(is, os, isClient);
    }
  
    /**
     * Initialize the multiplexor with input and output streams.
     *
     * @param is the underlying input stream
     * @param os the underlying output stream
     * @param isClient true if this is the connection client.
     */
    public void init(InputStream is, OutputStream os, boolean isClient)
    {
      this.is = is;
      this.os = os;
      this.isClient = isClient;
    }
  
    /**
     * Gets the raw input stream.  Clients will normally not call
     * this.
     */
    public InputStream getInputStream()
    {
      return is;
    }
  
    /**
     * Gets the raw output stream.  Clients will normally not call
     * this.
     */
    public OutputStream getOutputStream()
    {
      return os;
    }
  
    /**
     * Starts a client call.
     */
    public boolean startCall(MuxInputStream in, MuxOutputStream out)
      throws IOException
    {
      int channel = isClient ? 2 : 3;
  
      return startCall(channel, in, out);
    }
  
    /**
     * Starts a client call.
     */
    public boolean startCall(int channel, MuxInputStream in, MuxOutputStream out)
      throws IOException
    {
      // XXX: Eventually need to check to see if the channel is used.
      // It's not clear whether this should cause a wait or an exception.
      
      in.init(this, channel);
      out.init(this, channel);
  
      return true;
    }
    
    /**
     * Reads a server request.
     */
    public boolean readRequest(MuxInputStream in, MuxOutputStream out)
      throws IOException
    {
      int channel = isClient ? 3 : 2;
      
      in.init(this, channel);
      out.init(this, channel);
  
      if (readChannel(channel) != null) {
        in.setInputStream(is);
        in.readToData(false);
        return true;
      }
      else
        return false;
    }
  
    /**
     * Grabs the channel for writing.
     *
     * @param channel the channel
     *
     * @return true if the channel has permission to write.
     */
    OutputStream writeChannel(int channel)
      throws IOException
    {
      while (os != null) {
        boolean canWrite = false;
        synchronized (WRITE_LOCK) {
          if (! isWriteLocked) {
            isWriteLocked = true;
            canWrite = true;
          }
          else {
            try {
              WRITE_LOCK.wait(5000);
            } catch (Exception e) {
            }
          }
        }
  
        if (canWrite) {
          os.write('C');
          os.write(channel >> 8);
          os.write(channel);
          
          return os;
        }
      }
  
      return null;
    }
  
    void yield(int channel)
      throws IOException
    {
      os.write('Y');
      freeWriteLock();
    }
  
    void flush(int channel)
      throws IOException
    {
      os.write('Y');
      os.flush();
      freeWriteLock();
    }
  
    void close(int channel)
      throws IOException
    {
      if (os != null) {
        os.write('Q');
        os.flush();
        freeWriteLock();
      }
    }
    
    /**
     * Frees the channel for writing.
     */
    void freeWriteLock()
    {
      synchronized (WRITE_LOCK) {
        isWriteLocked = false;
        WRITE_LOCK.notifyAll();
      }
    }
  
    /**
     * Reads data from a channel.
     *
     * @param channel the channel
     *
     * @return true if the channel is valid.
     */
    InputStream readChannel(int channel)
      throws IOException
    {
      while (! isClosed) {
        if (inputReady[channel]) {
          inputReady[channel] = false;
          return is;
        }
  
        boolean canRead = false;
        synchronized (READ_LOCK) {
          if (! isReadLocked) {
            isReadLocked = true;
            canRead = true;
          }
          else {
            try {
              READ_LOCK.wait(5000);
            } catch (Exception e) {
            }
          }
        }
        
        if (canRead) {
          try {
            readData();
          } catch (IOException e) {
            close();
          }
        }
      }
  
      return null;
    }
  
    boolean getReadLock()
    {
      synchronized (READ_LOCK) {
        if (! isReadLocked) {
          isReadLocked = true;
          return true;
        }
        else {
          try {
            READ_LOCK.wait(5000);
          } catch (Exception e) {
          }
        }
      }
  
      return false;
    }
  
    /**
     * Frees the channel for reading.
     */
    void freeReadLock()
    {
      synchronized (READ_LOCK) {
        isReadLocked = false;
        READ_LOCK.notifyAll();
      }
    }
  
    /**
     * Reads data until a channel packet 'C' or error 'E' is received.
     */
    private void readData()
      throws IOException
    {
      while (! isClosed) {
        int code = is.read();
  
        switch (code) {
        case ' ':
        case '\t':
        case '\n':
        case '\r':
          break;
  
        case 'C': {
          int channel = (is.read() << 8) + is.read();
  
          inputReady[channel] = true;
          return;
        }
  
        case 'E': {
          int channel = (is.read() << 8) + is.read();
          int status = (is.read() << 8) + is.read();
  
          inputReady[channel] = true;
  
          return;
        }
  
        case -1:
          close();
          return;
          
        default:
          // An error in the protocol.  Kill the connection.
          close();
          return;
        }
      }
  
      return;
    }
  
    /**
     * Close the mux
     */
    public void close()
      throws IOException
    {
      isClosed = true;
      
      OutputStream os = this.os;
      this.os = null;
      
      InputStream is = this.is;
      this.is = null;
  
      if (os != null)
        os.close();
  
      if (is != null)
        is.close();
    }
  }
  