View Javadoc

    1   /*
    2    * Copyright 2009 Red Hat, Inc.
    3    *
    4    * Red Hat licenses this file to you under the Apache License, version 2.0
    5    * (the "License"); you may not use this file except in compliance with the
    6    * License.  You may obtain a copy of the License at:
    7    *
    8    *    http://www.apache.org/licenses/LICENSE-2.0
    9    *
    10   * Unless required by applicable law or agreed to in writing, software
    11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
    13   * License for the specific language governing permissions and limitations
    14   * under the License.
    15   */
    16  package org.jboss.netty.channel.local;
    17
    18  import static org.jboss.netty.channel.Channels.*;
    19
    20  import java.nio.channels.ClosedChannelException;
    21  import java.nio.channels.NotYetConnectedException;
    22  import java.util.Queue;
    23  import java.util.concurrent.atomic.AtomicBoolean;
    24
    25  import org.jboss.netty.channel.AbstractChannel;
    26  import org.jboss.netty.channel.ChannelConfig;
    27  import org.jboss.netty.channel.ChannelFactory;
    28  import org.jboss.netty.channel.ChannelFuture;
    29  import org.jboss.netty.channel.ChannelPipeline;
    30  import org.jboss.netty.channel.ChannelSink;
    31  import org.jboss.netty.channel.DefaultChannelConfig;
    32  import org.jboss.netty.channel.MessageEvent;
    33  import org.jboss.netty.util.internal.LinkedTransferQueue;
    34  import org.jboss.netty.util.internal.ThreadLocalBoolean;
    35
    36  /**
    37   * @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
    38   * @author Andy Taylor (andy.taylor@jboss.org)
    39   * @author <a href="http://gleamynode.net/">Trustin Lee</a>
    40   * @version $Rev: 2080 $, $Date: 2010-01-26 18:04:19 +0900 (Tue, 26 Jan 2010) $
    41   */
    42  final class DefaultLocalChannel extends AbstractChannel implements LocalChannel {
    43
    44      private final ChannelConfig config;
    45      private final ThreadLocalBoolean delivering = new ThreadLocalBoolean();
    46      final AtomicBoolean bound = new AtomicBoolean();
    47      final Queue<MessageEvent> writeBuffer = new LinkedTransferQueue<MessageEvent>();
    48
    49      volatile DefaultLocalChannel pairedChannel;
    50      volatile LocalAddress localAddress;
    51      volatile LocalAddress remoteAddress;
    52
    53      DefaultLocalChannel(LocalServerChannel parent, ChannelFactory factory, ChannelPipeline pipeline, ChannelSink sink, DefaultLocalChannel pairedChannel) {
    54          super(parent, factory, pipeline, sink);
    55          this.pairedChannel = pairedChannel;
    56          config = new DefaultChannelConfig();
    57          fireChannelOpen(this);
    58      }
    59
    60      public ChannelConfig getConfig() {
    61          return config;
    62      }
    63
    64      public boolean isBound() {
    65          return bound.get() && isOpen();
    66      }
    67
    68      public boolean isConnected() {
    69          return pairedChannel != null && isOpen();
    70      }
    71
    72      public LocalAddress getLocalAddress() {
    73          return localAddress;
    74      }
    75
    76      public LocalAddress getRemoteAddress() {
    77          return remoteAddress;
    78      }
    79
    80      void closeNow(ChannelFuture future) {
    81          LocalAddress localAddress = this.localAddress;
    82          try {
    83              // Close the self.
    84              if (!setClosed()) {
    85                  return;
    86              }
    87
    88              DefaultLocalChannel pairedChannel = this.pairedChannel;
    89              if (pairedChannel != null) {
    90                  this.pairedChannel = null;
    91                  fireChannelDisconnected(this);
    92                  fireChannelUnbound(this);
    93              }
    94              fireChannelClosed(this);
    95
    96              // Close the peer.
    97              if (pairedChannel == null || !pairedChannel.setClosed()) {
    98                  return;
    99              }
    100
    101             DefaultLocalChannel me = pairedChannel.pairedChannel;
    102             if (me != null) {
    103                 pairedChannel.pairedChannel = null;
    104                 fireChannelDisconnected(pairedChannel);
    105                 fireChannelUnbound(pairedChannel);
    106             }
    107             fireChannelClosed(pairedChannel);
    108         } finally {
    109             future.setSuccess();
    110             if (localAddress != null && getParent() == null) {
    111                 LocalChannelRegistry.unregister(localAddress);
    112             }
    113         }
    114     }
    115
    116     void flushWriteBuffer() {
    117         DefaultLocalChannel pairedChannel = this.pairedChannel;
    118         if (pairedChannel != null) {
    119             if (pairedChannel.isConnected()){
    120                 // Channel is open and connected and channelConnected event has
    121                 // been fired.
    122                 if (!delivering.get()) {
    123                     delivering.set(true);
    124                     try {
    125                         for (;;) {
    126                             MessageEvent e = writeBuffer.poll();
    127                             if(e == null) {
    128                                 break;
    129                             }
    130
    131                             e.getFuture().setSuccess();
    132                             fireMessageReceived(pairedChannel, e.getMessage());
    133                             fireWriteComplete(this, 1);
    134                         }
    135                     } finally {
    136                         delivering.set(false);
    137                     }
    138                 }
    139             } else {
    140                 // Channel is open and connected but channelConnected event has
    141                 // not been fired yet.
    142             }
    143         } else {
    144             // Channel is closed or not connected yet - notify as failures.
    145             Exception cause;
    146             if (isOpen()) {
    147                 cause = new NotYetConnectedException();
    148             } else {
    149                 cause = new ClosedChannelException();
    150             }
    151
    152             for (;;) {
    153                 MessageEvent e = writeBuffer.poll();
    154                 if(e == null) {
    155                     break;
    156                 }
    157
    158                 e.getFuture().setFailure(cause);
    159                 fireExceptionCaught(this, cause);
    160             }
    161         }
    162     }
    163 }