    1   /***
    2    *
    3    * Copyright 2004 Hiram Chirino
    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  import java.net.URI;
    23
    24  import org.activeio.Channel;
    25  import org.activeio.FilterSynchChannel;
    26  import org.activeio.FilterSynchChannelServer;
    27  import org.activeio.Packet;
    28  import org.activeio.SynchChannel;
    29  import org.activeio.SynchChannelFactory;
    30  import org.activeio.SynchChannelServer;
    31
    32  /***
    33   * Makes all the channels produced by another [@see org.activeio.SynchChannelFactory}
    34   * have write operations that have built in delays for testing.
    35   *
    36   * @version $Revision$
    37   */
    38  public class SlowWriteSynchChannelFactory implements SynchChannelFactory {
    39
    40      final SynchChannelFactory next;
    41      private final int maxPacketSize;
    42      private final long packetDelay;
    43
    44      public SlowWriteSynchChannelFactory(final SynchChannelFactory next, int maxPacketSize, long packetDelay) {
    45          this.next = next;
    46          this.maxPacketSize = maxPacketSize;
    47          this.packetDelay = packetDelay;
    48      }
    49
    50      class SlowWriteSynchChannel extends FilterSynchChannel {
    51          public SlowWriteSynchChannel(SynchChannel next) {
    52              super(next);
    53          }
    54          public void write(Packet packet) throws IOException {
    55              packet = packet.slice();
    56              while(packet.hasRemaining()) {
    57                  int size = Math.max(maxPacketSize, packet.remaining());
    58                  packet.position(size);
    59                  Packet remaining = packet.slice();
    60                  packet.flip();
    61                  Packet data = packet.slice();
    62                  super.write(data);
    63                  packet = remaining;
    64                  try {
    65                      Thread.sleep(packetDelay);
    66                  } catch (InterruptedException e) {
    67                      throw new InterruptedIOException();
    68                  }
    69              }
    70          }
    71      }
    72
    73      class SlowWriteSynchChannelServer extends FilterSynchChannelServer {
    74          public SlowWriteSynchChannelServer(SynchChannelServer next) {
    75              super(next);
    76          }
    77          public Channel accept(long timeout) throws IOException {
    78              Channel channel = super.accept(timeout);
    79              if( channel != null ) {
    80                  channel =  new SlowWriteSynchChannel((SynchChannel) channel);
    81              }
    82              return channel;
    83          }
    84      }
    85
    86      public SynchChannelServer bindSynchChannel(URI location) throws IOException {
    87          return next.bindSynchChannel(location);
    88      }
    89
    90      public SynchChannel openSynchChannel(URI location) throws IOException {
    91          return new SlowWriteSynchChannel(next.openSynchChannel(location));
    92      }
    93
    94  }