View Javadoc

    1   /***
    2    *
    3    * Copyright 2004 Protique Ltd
    4    *
    5    * Licensed under the Apache License, Version 2.0 (the "License");
    6    * you may not use this file except in compliance with the License.
    7    * You may obtain a copy of the License at
    8    *
    9    * http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   **/
    18  package org.activeio.net;
    19
    20  import java.io.IOException;
    21  import java.io.InterruptedIOException;
    22
    23  import org.activeio.AsynchChannel;
    24  import org.activeio.AsynchChannelListener;
    25  import org.activeio.Packet;
    26  import org.activeio.packet.EOSPacket;
    27
    28  import EDU.oswego.cs.dl.util.concurrent.Semaphore;
    29
    30  /***
    31   * Used to connect the bottom ends of two Asynch channel stacks.
    32   *
    33   */
    34  final public class VMPipeAsynchChannelPipe {
    35
    36      final PipeChannel leftChannel = new PipeChannel();
    37      final PipeChannel rightChannel = new PipeChannel();
    38
    39      final public static class PipeChannel implements AsynchChannel {
    40
    41          private PipeChannel sibiling;
    42          private AsynchChannelListener channelListener;
    43          private final Semaphore runMutext = new Semaphore(0);
    44          private boolean disposed;
    45          private boolean running;
    46
    47          public PipeChannel() {
    48          }
    49
    50          public void setAsynchChannelListener(AsynchChannelListener channelListener) {
    51              this.channelListener = channelListener;
    52          }
    53          public AsynchChannelListener getAsynchChannelListener() {
    54              return channelListener;
    55          }
    56
    57          public void write(Packet packet) throws IOException {
    58              if( disposed )
    59                  throw new IOException("Conneciton closed.");
    60              sibiling.onPacket(packet, WAIT_FOREVER_TIMEOUT);
    61          }
    62
    63          private void onPacket(Packet packet, long timeout) throws IOException {
    64              try {
    65                  if( timeout == NO_WAIT_TIMEOUT ) {
    66                      if( !runMutext.attempt(0) )
    67                          return;
    68                  } else if( timeout == WAIT_FOREVER_TIMEOUT ) {
    69                      runMutext.acquire();
    70                  } else {
    71                      if( !runMutext.attempt(timeout) )
    72                          return;
    73                  }
    74              } catch (InterruptedException e) {
    75                  throw new InterruptedIOException();
    76              }
    77              try {
    78                  if( disposed ) {
    79                      throw new IOException("Peer connection closed.");
    80                  }
    81                  channelListener.onPacket(packet);
    82              } finally {
    83                  runMutext.release();
    84              }
    85          }
    86
    87          public void flush() throws IOException {
    88          }
    89
    90          public void start() throws IOException {
    91              if(running)
    92                  return;
    93              if( channelListener==null )
    94                  throw new IOException("channelListener has not been set.");
    95              running=true;
    96              runMutext.release();
    97          }
    98
    99          public void stop(long timeout) throws IOException {
    100             if(!running)
    101                 return;
    102             try {
    103                 if( timeout == NO_WAIT_TIMEOUT ) {
    104                     if( !runMutext.attempt(0) )
    105                         return;
    106                 } else if( timeout == WAIT_FOREVER_TIMEOUT ) {
    107                     runMutext.acquire();
    108                 } else {
    109                     if( !runMutext.attempt(timeout) )
    110                         return;
    111                 }
    112                 running=false;
    113             } catch (InterruptedException e) {
    114                 throw new InterruptedIOException();
    115             }
    116         }
    117
    118         public void dispose() {
    119             if( disposed )
    120                 return;
    121
    122             if( running && channelListener!=null ) {
    123                 this.channelListener.onPacketError(new IOException("Pipe closed."));
    124                 running=false;
    125             }
    126             disposed = true;
    127             runMutext.release();
    128
    129             try {
    130                 // Inform the peer of the End Of Stream if he's listening.
    131                 sibiling.onPacket(EOSPacket.EOS_PACKET, NO_WAIT_TIMEOUT);
    132             } catch (IOException e) {
    133             }
    134         }
    135
    136         public PipeChannel getSibiling() {
    137             return sibiling;
    138         }
    139         public void setSibiling(PipeChannel sibiling) {
    140             this.sibiling = sibiling;
    141         }
    142
    143         public Object narrow(Class target) {
    144             if( target.isAssignableFrom(getClass()) ) {
    145                 return this;
    146             }
    147             return null;
    148         }
    149
    150         public String getId() {
    151             return "0x"+Integer.toHexString(System.identityHashCode(this));
    152         }
    153
    154         public String toString() {
    155             return "Pipe Channel from "+getId()+" to "+sibiling.getId();
    156         }
    157     }
    158
    159     public VMPipeAsynchChannelPipe() {
    160         leftChannel.setSibiling(rightChannel);
    161         rightChannel.setSibiling(leftChannel);
    162     }
    163
    164     public AsynchChannel getLeftAsynchChannel() {
    165         return leftChannel;
    166     }
    167
    168     public AsynchChannel getRightAsynchChannel() {
    169         return rightChannel;
    170     }
    171 }