View Javadoc

    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  package org.activeio.adapter;
    18
    19  import java.io.IOException;
    20  import java.io.InterruptedIOException;
    21
    22  import org.activeio.AsynchChannel;
    23  import org.activeio.ChannelFactory;
    24  import org.activeio.FilterAsynchChannel;
    25  import org.activeio.Packet;
    26  import org.activeio.PacketData;
    27  import org.activeio.RequestChannel;
    28  import org.activeio.RequestListener;
    29  import org.activeio.packet.AppendedPacket;
    30  import org.activeio.packet.ByteArrayPacket;
    31
    32  import EDU.oswego.cs.dl.util.concurrent.ConcurrentHashMap;
    33  import EDU.oswego.cs.dl.util.concurrent.Executor;
    34  import EDU.oswego.cs.dl.util.concurrent.Slot;
    35
    36
    37  /***
    38   * Creates a {@see org.activeio.RequestChannel} out of a {@see org.activeio.AsynchChannel}.  This
    39   * {@see org.activeio.RequestChannel} is thread safe and mutiplexes concurrent requests and responses over
    40   * the underlying {@see org.activeio.AsynchChannel}.
    41   *
    42   * @version $Revision$
    43   */
    44  final public class AsynchChannelToConcurrentRequestChannel extends FilterAsynchChannel implements RequestChannel {
    45
    46      private static final byte PASSTHROUGH = 0x00;
    47      private static final byte REQUEST = 0x01;
    48      private static final byte RESPONSE = 0x02;
    49      private static final ByteArrayPacket PASSTHROUGH_PACKET = new ByteArrayPacket(new byte[]{PASSTHROUGH});
    50
    51      private final ConcurrentHashMap requestMap = new ConcurrentHashMap();
    52      private final Executor requestExecutor;
    53      private short nextRequestId = 0;
    54      private final Object writeMutex = new Object();
    55
    56      private RequestListener requestListener;
    57
    58      public AsynchChannelToConcurrentRequestChannel(AsynchChannel next) {
    59          this(next, ChannelFactory.DEFAULT_EXECUTOR);
    60      }
    61
    62      public AsynchChannelToConcurrentRequestChannel(AsynchChannel next, Executor requestExecutor) {
    63          super(next);
    64          this.requestExecutor=requestExecutor;
    65      }
    66
    67      synchronized short getNextRequestId() {
    68          return nextRequestId++;
    69      }
    70
    71      /***
    72       * @see org.activeio.FilterAsynchChannel#write(org.activeio.channel.Packet)
    73       */
    74      public void write(Packet packet) throws IOException {
    75          Packet passThrough = AppendedPacket.join(PASSTHROUGH_PACKET.duplicate(), packet);
    76          synchronized(writeMutex) {
    77              super.write(passThrough);
    78          }
    79      }
    80
    81      /***
    82       * @see org.activeio.FilterAsynchChannel#onPacket(org.activeio.channel.Packet)
    83       */
    84      public void onPacket(final Packet packet) {
    85              switch( packet.read() ) {
    86              	case PASSTHROUGH:
    87                      super.onPacket(packet);
    88                      break;
    89              	case REQUEST:
    90                      try {
    91  	            	    requestExecutor.execute(new Runnable(){
    92  	                        public void run() {
    93  	                    	    serviceRequest(packet);
    94  	                        }
    95  	                    });
    96                      } catch (InterruptedException e) {
    97                          Thread.currentThread().interrupt();
    98                      }
    99              	    break;
    100             	case RESPONSE:
    101                     serviceReponse(packet);
    102             	    break;
    103             }
    104     }
    105
    106     private void serviceRequest(Packet packet) {
    107         try {
    108             if( requestListener ==null )
    109                 throw new IOException("The RequestListener has not been set.");
    110
    111             PacketData data = new PacketData(packet);
    112             short requestId = data.readShort();
    113             Packet reponse = requestListener.onRequest(packet);
    114
    115             // Send the response...
    116             Packet header = createHeaderPacket(RESPONSE, requestId);
    117             Packet rc = AppendedPacket.join(header, packet);
    118             synchronized(writeMutex) {
    119                 super.write(rc);
    120             }
    121         } catch (IOException e) {
    122             super.onPacketError(e);
    123         }
    124
    125     }
    126
    127     private void serviceReponse(Packet packet) {
    128
    129         try {
    130
    131             PacketData data = new PacketData(packet);
    132             short requestId = data.readShort();
    133
    134             Slot responseSlot = (Slot) requestMap.get(new Short(requestId));
    135             responseSlot.put(packet);
    136
    137         } catch (IOException e) {
    138             super.onPacketError(e);
    139         } catch (InterruptedException e) {
    140             Thread.currentThread().interrupt();
    141         }
    142
    143
    144     }
    145
    146     public Packet request(Packet request, long timeout) throws IOException {
    147
    148         Short requestId = new Short(getNextRequestId());
    149         Slot responseSlot = new Slot();
    150         requestMap.put(requestId, responseSlot);
    151
    152         Packet header = createHeaderPacket(REQUEST, requestId.shortValue());
    153         Packet packet = AppendedPacket.join(header, request);
    154
    155         synchronized(writeMutex) {
    156             super.write(packet);
    157         }
    158
    159         try {
    160
    161             if( timeout == WAIT_FOREVER_TIMEOUT ) {
    162                 return (Packet) responseSlot.take();
    163             } else if (timeout == NO_WAIT_TIMEOUT ) {
    164                 return (Packet) responseSlot.poll(1);
    165             } else {
    166                 return (Packet) responseSlot.poll(timeout);
    167             }
    168
    169         } catch (InterruptedException e) {
    170             throw new InterruptedIOException(e.getMessage());
    171         } finally {
    172             requestMap.remove(requestId);
    173         }
    174     }
    175
    176     private Packet createHeaderPacket(byte type, short requestId) throws IOException {
    177         ByteArrayPacket header = new ByteArrayPacket(new byte[]{3});
    178         PacketData data = new PacketData(header);
    179         data.writeByte(type);
    180         data.writeShort(requestId);
    181         header.flip();
    182         return header;
    183     }
    184
    185     public void setRequestListener(RequestListener requestListener) throws IOException {
    186         this.requestListener = requestListener;
    187     }
    188
    189     public RequestListener getRequestListener() {
    190         return requestListener;
    191     }
    192 }