protocol.java

用java实现的一个socket服务器。采用非阻塞模式

  public void stop()
  {
    stopInternal();
  }

  /**
   * This method is called on a {@link org.jgroups.Channel#close()}.
   * Does some cleanup; after the call the VM will terminate
   */
  public void destroy()
  {
  }

  public Queue getUpQueue()
  {
    return up_queue;
  } // used by Debugger (ProtocolView)

  public Queue getDownQueue()
  {
    return down_queue;
  } // used by Debugger (ProtocolView)

  /** List of events that are required to be answered by some layer above.
   @return Vector (of Integers) */
  public Vector requiredUpServices()
  {
    return null;
  }

  /** List of events that are required to be answered by some layer below.
   @return Vector (of Integers) */
  public Vector requiredDownServices()
  {
    return null;
  }

  /** List of events that are provided to layers above (they will be handled when sent down from
   above).
   @return Vector (of Integers) */
  public Vector providedUpServices()
  {
    return null;
  }

  /** List of events that are provided to layers below (they will be handled when sent down from
   below).
   @return Vector (of Integers) */
  public Vector providedDownServices()
  {
    return null;
  }

  public abstract String getName(); // all protocol names have to be unique !

  public Protocol getUpProtocol()
  {
    return up_prot;
  }

  public Protocol getDownProtocol()
  {
    return down_prot;
  }

  public void setUpProtocol(Protocol up_prot)
  {
    this.up_prot = up_prot;
  }

  public void setDownProtocol(Protocol down_prot)
  {
    this.down_prot = down_prot;
  }

  public void setProtocolStack(ProtocolStack stack)
  {
    this.stack = stack;
  }

  /** Used internally. If overridden, call this method first. Only creates the up_handler thread
   if down_thread is true */
  public void startUpHandler()
  {
    if (up_thread)
    {
      if (up_handler == null)
      {
        up_handler = new UpHandler(this, up_queue, this, observer);
        if (up_thread_prio >= 0)
        {
          try
          {
            up_handler.setPriority(up_thread_prio);
          }
          catch (Throwable t)
          {
            if (log.isErrorEnabled())
            {
              log.error("priority " + up_thread_prio +
                        " could not be set for thread", t);
            }
          }
        }
        up_handler.start();
      }
    }
  }

  /** Used internally. If overridden, call this method first. Only creates the down_handler thread
   if down_thread is true */
  public void startDownHandler()
  {
    if (down_thread)
    {
      if (down_handler == null)
      {
        down_handler = new DownHandler(this, down_queue, this, observer);
        if (down_thread_prio >= 0)
        {
          try
          {
            down_handler.setPriority(down_thread_prio);
          }
          catch (Throwable t)
          {
            if (log.isErrorEnabled())
            {
              log.error("priority " + down_thread_prio +
                        " could not be set for thread", t);
            }
          }
        }
        down_handler.start();
      }
    }
  }

  /** Used internally. If overridden, call parent's method first */
  public void stopInternal()
  {
    up_queue.close(false); // this should terminate up_handler thread

    if (up_handler != null && up_handler.isAlive())
    {
      try
      {
        up_handler.join(THREAD_JOIN_TIMEOUT);
      }
      catch (Exception ex)
      {
      }
      if (up_handler != null && up_handler.isAlive())
      {
        up_handler.interrupt(); // still alive ? let's just kill it without mercy...
        try
        {
          up_handler.join(THREAD_JOIN_TIMEOUT);
        }
        catch (Exception ex)
        {
        }
        if (up_handler != null && up_handler.isAlive())
        {
          if (log.isErrorEnabled())
          {
            log.error("up_handler thread for " + getName() +
                      " was interrupted (in order to be terminated), but is still alive");
          }
        }
      }
    }
    up_handler = null;

    down_queue.close(false); // this should terminate down_handler thread
    if (down_handler != null && down_handler.isAlive())
    {
      try
      {
        down_handler.join(THREAD_JOIN_TIMEOUT);
      }
      catch (Exception ex)
      {
      }
      if (down_handler != null && down_handler.isAlive())
      {
        down_handler.interrupt(); // still alive ? let's just kill it without mercy...
        try
        {
          down_handler.join(THREAD_JOIN_TIMEOUT);
        }
        catch (Exception ex)
        {
        }
        if (down_handler != null && down_handler.isAlive())
        {
          if (log.isErrorEnabled())
          {
            log.error("down_handler thread for " + getName() +
                      " was interrupted (in order to be terminated), but is is still alive");
          }
        }
      }
    }
    down_handler = null;
  }

  public Event handleUpEvent(Event evt)
  {

    return evt;
  }

  public Event handleDownEvent(Event evt)
  {

    return evt;
  }

  /**
   * Internal method, should not be called by clients. Used by ProtocolStack. I would have
   * used the 'friends' modifier, but this is available only in C++ ... If the up_handler thread
   * is not available (down_thread == false), then directly call the up() method: we will run on the
   * caller's thread (e.g. the protocol layer below us).
   */
  protected void receiveUpEvent(Event evt)
  {
    if (up_handler == null)
    {
      if (observer != null)
      { // call debugger hook (if installed)
        if (observer.up(evt, up_queue.size()) == false)
        { // false means discard event
          return;
        }
      }
      evt = handleUpEvent(evt);
      if (null != evt)
      {
        up(evt);
      }
      return;
    }
    try
    {
      up_queue.add(evt);
    }
    catch (Exception e)
    {
      if (log.isWarnEnabled())
      {
        log.warn("exception: " + e);
      }
    }
  }

  /**
   * Internal method, should not be called by clients. Used by ProtocolStack. I would have
   * used the 'friends' modifier, but this is available only in C++ ... If the down_handler thread
   * is not available (down_thread == false), then directly call the down() method: we will run on the
   * caller's thread (e.g. the protocol layer above us).
   */
  protected void receiveDownEvent(Event evt)
  {
    if (down_handler == null)
    {
      if (observer != null)
      { // call debugger hook (if installed)
        if (observer.down(evt, down_queue.size()) == false)
        { // false means discard event
          return;
        }
      }
      int type = evt.getType();
      if (type == Event.START || type == Event.STOP)
      {
        if (handleSpecialDownEvent(evt) == false)
        {
          return;
        }
      }
      evt = handleDownEvent(evt);
      if (null != evt)
      {
        down(evt);
      }

      return;
    }
    try
    {
      down_queue.add(evt);
    }
    catch (Exception e)
    {
      if (log.isWarnEnabled())
      {
        log.warn("exception: " + e);
      }
    }
  }

  /**
   * Causes the event to be forwarded to the next layer up in the hierarchy. Typically called
   * by the implementation of <code>Up</code> (when done).
   */
  public void passUp(Event evt)
  {
    if (observer != null)
    { // call debugger hook (if installed)
      if (observer.passUp(evt) == false)
      { // false means don't pass up (=discard) event
        return;
      }
    }
    up_prot.receiveUpEvent(evt);
  }

  /**
   * Causes the event to be forwarded to the next layer down in the hierarchy.Typically called
   * by the implementation of <code>Down</code> (when done).
   */
  public void passDown(Event evt)
  {
    if (observer != null)
    { // call debugger hook (if installed)
      if (observer.passDown(evt) == false)
      { // false means don't pass down (=discard) event
        return;
      }
    }
    down_prot.receiveDownEvent(evt);
  }

  /**
   * An event was received from the layer below. Usually the current layer will want to examine
   * the event type and - depending on its type - perform some computation
   * (e.g. removing headers from a MSG event type, or updating the internal membership list
   * when receiving a VIEW_CHANGE event).
   * Finally the event is either a) discarded, or b) an event is sent down
   * the stack using <code>passDown()</code> or c) the event (or another event) is sent up
   * the stack using <code>passUp()</code>.
   */
  public void up(Event evt)
  {
    passUp(evt);
  }

  /**
   * An event is to be sent down the stack. The layer may want to examine its type and perform
   * some action on it, depending on the event's type. If the event is a message MSG, then
   * the layer may need to add a header to it (or do nothing at all) before sending it down
   * the stack using <code>passDown()</code>. In case of a GET_ADDRESS event (which tries to
   * retrieve the stack's address from one of the bottom layers), the layer may need to send
   * a new response event back up the stack using <code>passUp()</code>.
   */
  public void down(Event evt)
  {
    passDown(evt);
  }

  /**  These are special internal events that should not be handled by protocols
   * @return boolean True: the event should be passed further down the stack. False: the event should
   * be discarded (not passed down the stack)
   */
  protected boolean handleSpecialDownEvent(Event evt)
  {
    switch (evt.getType())
    {
      case Event.START:
        try
        {
          start();

          // if we're the transport protocol, reply with a START_OK up the stack
          if (down_prot == null)
          {
            passUp(new Event(Event.START_OK, Boolean.TRUE));
            return false; // don't pass down the stack
          }
          else
          {
            return true; // pass down the stack
          }
        }
        catch (Exception e)
        {
          passUp(new Event(Event.START_OK,
                           new Exception("exception caused by " + getName() +
                                         ".start()", e)));
          return false;
        }
        case Event.STOP:
          stop();
          if (down_prot == null)
          {
            passUp(new Event(Event.STOP_OK, Boolean.TRUE));
            return false; // don't pass down the stack
          }
          else
          {
            return true; // pass down the stack
          }
      default:
        return true; // pass down by default
    }
  }
}