View Javadoc

    1   /*
    2    *  Licensed to the Apache Software Foundation (ASF) under one
    3    *  or more contributor license agreements.  See the NOTICE file
    4    *  distributed with this work for additional information
    5    *  regarding copyright ownership.  The ASF licenses this file
    6    *  to you under the Apache License, Version 2.0 (the
    7    *  "License"); you may not use this file except in compliance
    8    *  with the License.  You may obtain a copy of the License at
    9    *
    10   *    http://www.apache.org/licenses/LICENSE-2.0
    11   *
    12   *  Unless required by applicable law or agreed to in writing,
    13   *  software distributed under the License is distributed on an
    14   *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    15   *  KIND, either express or implied.  See the License for the
    16   *  specific language governing permissions and limitations
    17   *  under the License.
    18   *
    19   */
    20  package org.apache.mina.transport.vmpipe.support;
    21
    22  import java.util.Queue;
    23  import java.util.concurrent.ConcurrentLinkedQueue;
    24
    25  import org.apache.mina.common.ByteBuffer;
    26  import org.apache.mina.common.IdleStatus;
    27  import org.apache.mina.common.IoSession;
    28  import org.apache.mina.common.IoFilter.WriteRequest;
    29  import org.apache.mina.common.support.AbstractIoFilterChain;
    30
    31  /**
    32   * @todo Document me!
    33   *
    34   * @author The Apache Directory Project (mina-dev@directory.apache.org)
    35   * @version $Rev: 585081 $, $Date: 2007-10-16 10:45:14 +0200 (Tue, 16 Oct 2007) $
    36   */
    37  public class VmPipeFilterChain extends AbstractIoFilterChain {
    38
    39      private final Queue<Event> eventQueue = new ConcurrentLinkedQueue<Event>();
    40
    41      private volatile boolean flushEnabled;
    42      private volatile boolean sessionOpened;
    43
    44      public VmPipeFilterChain(IoSession session) {
    45          super(session);
    46      }
    47
    48      public void start() {
    49          flushEnabled = true;
    50          flushEvents();
    51          flushPendingDataQueues( (VmPipeSessionImpl) getSession() );
    52      }
    53
    54      private void pushEvent(Event e) {
    55          eventQueue.offer(e);
    56          if ( flushEnabled ) {
    57              flushEvents();
    58          }
    59      }
    60
    61      private void flushEvents() {
    62          Event e;
    63          while ((e = eventQueue.poll()) != null) {
    64              fireEvent(e);
    65          }
    66      }
    67
    68      private void fireEvent(Event e) {
    69          IoSession session = getSession();
    70          EventType type = e.getType();
    71          Object data = e.getData();
    72
    73          if (type == EventType.RECEIVED) {
    74              VmPipeSessionImpl s = (VmPipeSessionImpl) session;
    75
    76              if( sessionOpened && s.getTrafficMask().isReadable() && s.getLock().tryLock()) {
    77                  try {
    78                      int byteCount = 1;
    79                      if (data instanceof ByteBuffer) {
    80                          byteCount = ((ByteBuffer) data).remaining();
    81                      }
    82
    83                      s.increaseReadBytes(byteCount);
    84
    85                      super.fireMessageReceived(s, data);
    86                  } finally {
    87                      s.getLock().unlock();
    88                  }
    89
    90                  flushPendingDataQueues( s );
    91              } else {
    92                  s.pendingDataQueue.add(data);
    93              }
    94          } else if (type == EventType.WRITE) {
    95              super.fireFilterWrite(session, (WriteRequest) data);
    96          } else if (type == EventType.SENT) {
    97              super.fireMessageSent(session, (WriteRequest) data);
    98          } else if (type == EventType.EXCEPTION) {
    99              super.fireExceptionCaught(session, (Throwable) data);
    100         } else if (type == EventType.IDLE) {
    101             super.fireSessionIdle(session, (IdleStatus) data);
    102         } else if (type == EventType.OPENED) {
    103             super.fireSessionOpened(session);
    104             sessionOpened = true;
    105         } else if (type == EventType.CREATED) {
    106             super.fireSessionCreated(session);
    107         } else if (type == EventType.CLOSED) {
    108             super.fireSessionClosed(session);
    109         } else if (type == EventType.CLOSE) {
    110             super.fireFilterClose(session);
    111         }
    112     }
    113
    114     private static void flushPendingDataQueues( VmPipeSessionImpl s ) {
    115         s.updateTrafficMask();
    116         s.getRemoteSession().updateTrafficMask();
    117     }
    118
    119     @Override
    120     public void fireFilterClose(IoSession session) {
    121         pushEvent(new Event(EventType.CLOSE, null));
    122     }
    123
    124     @Override
    125     public void fireFilterWrite(IoSession session, WriteRequest writeRequest) {
    126         pushEvent(new Event(EventType.WRITE, writeRequest));
    127     }
    128
    129     @Override
    130     public void fireExceptionCaught(IoSession session, Throwable cause) {
    131         pushEvent(new Event(EventType.EXCEPTION, cause));
    132     }
    133
    134     @Override
    135     public void fireMessageSent(IoSession session, WriteRequest request) {
    136         pushEvent(new Event(EventType.SENT, request));
    137     }
    138
    139     @Override
    140     public void fireSessionClosed(IoSession session) {
    141         pushEvent(new Event(EventType.CLOSED, null));
    142     }
    143
    144     @Override
    145     public void fireSessionCreated(IoSession session) {
    146         pushEvent(new Event(EventType.CREATED, null));
    147     }
    148
    149     @Override
    150     public void fireSessionIdle(IoSession session, IdleStatus status) {
    151         pushEvent(new Event(EventType.IDLE, status));
    152     }
    153
    154     @Override
    155     public void fireSessionOpened(IoSession session) {
    156         pushEvent(new Event(EventType.OPENED, null));
    157     }
    158
    159     @Override
    160     public void fireMessageReceived(IoSession session, Object message) {
    161         pushEvent(new Event(EventType.RECEIVED, message));
    162     }
    163
    164     @Override
    165     protected void doWrite(IoSession session, WriteRequest writeRequest) {
    166         VmPipeSessionImpl s = (VmPipeSessionImpl) session;
    167         if (s.isConnected()) {
    168             if ( s.getTrafficMask().isWritable() && s.getLock().tryLock()) {
    169                 try {
    170                     Object message = writeRequest.getMessage();
    171
    172                     int byteCount = 1;
    173                     Object messageCopy = message;
    174                     if (message instanceof ByteBuffer) {
    175                         ByteBuffer rb = (ByteBuffer) message;
    176                         rb.mark();
    177                         byteCount = rb.remaining();
    178                         ByteBuffer wb = ByteBuffer.allocate(rb.remaining());
    179                         wb.put(rb);
    180                         wb.flip();
    181                         rb.reset();
    182                         messageCopy = wb;
    183                     }
    184
    185                     // Avoid unwanted side effect that scheduledWrite* becomes negative
    186                     // by increasing them.
    187                     s.increaseScheduledWriteBytes(byteCount);
    188                     s.increaseScheduledWriteRequests();
    189
    190                     s.increaseWrittenBytes(byteCount);
    191                     s.increaseWrittenMessages();
    192
    193                     s.getRemoteSession().getFilterChain().fireMessageReceived(
    194                         s.getRemoteSession(), messageCopy);
    195                     s.getFilterChain().fireMessageSent(s, writeRequest);
    196                 } finally {
    197                     s.getLock().unlock();
    198                 }
    199
    200                 flushPendingDataQueues( s );
    201             } else {
    202                 s.pendingDataQueue.add(writeRequest);
    203             }
    204         } else {
    205             writeRequest.getFuture().setWritten(false);
    206         }
    207     }
    208
    209     @Override
    210     protected void doClose(IoSession session) {
    211         VmPipeSessionImpl s = (VmPipeSessionImpl) session;
    212
    213         try {
    214             s.getLock().lock();
    215
    216             if (!session.getCloseFuture().isClosed()) {
    217                 s.getServiceListeners().fireSessionDestroyed(s);
    218                 s.getRemoteSession().close();
    219             }
    220         } finally {
    221             s.getLock().unlock();
    222         }
    223     }
    224
    225     // FIXME Copied and pasted from {@link ExecutorFilter}.
    226     private static class EventType {
    227         public static final EventType CREATED = new EventType("CREATED");
    228
    229         public static final EventType OPENED = new EventType("OPENED");
    230
    231         public static final EventType CLOSED = new EventType("CLOSED");
    232
    233         public static final EventType RECEIVED = new EventType("RECEIVED");
    234
    235         public static final EventType SENT = new EventType("SENT");
    236
    237         public static final EventType IDLE = new EventType("IDLE");
    238
    239         public static final EventType EXCEPTION = new EventType("EXCEPTION");
    240
    241         public static final EventType WRITE = new EventType("WRITE");
    242
    243         public static final EventType CLOSE = new EventType("CLOSE");
    244
    245         private final String value;
    246
    247         private EventType(String value) {
    248             this.value = value;
    249         }
    250
    251         @Override
    252         public String toString() {
    253             return value;
    254         }
    255     }
    256
    257     private static class Event {
    258         private final EventType type;
    259
    260         private final Object data;
    261
    262         private Event(EventType type, Object data) {
    263             this.type = type;
    264             this.data = data;
    265         }
    266
    267         public Object getData() {
    268             return data;
    269         }
    270
    271         public EventType getType() {
    272             return type;
    273         }
    274     }
    275 }