1   /**
    2    * Licensed to the Apache Software Foundation (ASF) under one or more
    3    * contributor license agreements.  See the NOTICE file distributed with
    4    * this work for additional information regarding copyright ownership.
    5    * The ASF licenses this file to You under the Apache License, Version 2.0
    6    * (the "License"); you may not use this file except in compliance with
    7    * the License.  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.apache.activemq.transport.nio;
   18   
   19   import java.nio.channels.CancelledKeyException;
   20   import java.nio.channels.ClosedChannelException;
   21   import java.nio.channels.SelectionKey;
   22   import java.nio.channels.SocketChannel;
   23   import java.util.concurrent.atomic.AtomicBoolean;
   24   
   25   import org.apache.activemq.transport.nio.SelectorManager.Listener;
   26   
   27   /**
   28    * @author chirino
   29    */
   30   public final class SelectorSelection {
   31   
   32       private final SelectorWorker worker;
   33       private final Listener listener;
   34       private int interest;
   35       private SelectionKey key;
   36       private AtomicBoolean closed = new AtomicBoolean();
   37   
   38       public SelectorSelection(final SelectorWorker worker, final SocketChannel socketChannel, Listener listener) throws ClosedChannelException {
   39           this.worker = worker;
   40           this.listener = listener;
   41           worker.addIoTask(new Runnable() {
   42               public void run() {
   43                   try {
   44                       SelectorSelection.this.key = socketChannel.register(worker.selector, 0, SelectorSelection.this);
   45                   } catch (Exception e) {
   46                       e.printStackTrace();
   47                   }
   48               }
   49           });
   50       }
   51   
   52       public void setInterestOps(int ops) {
   53           interest = ops;
   54       }
   55   
   56       public void enable() {
   57           worker.addIoTask(new Runnable() {
   58               public void run() {
   59                   try {
   60                       key.interestOps(interest);
   61                   } catch (CancelledKeyException e) {
   62                   }
   63               }
   64           });        
   65       }
   66   
   67       public void disable() {
   68           worker.addIoTask(new Runnable() {
   69               public void run() {
   70                   try {
   71                       key.interestOps(0);
   72                   } catch (CancelledKeyException e) {
   73                   }
   74               }
   75           });        
   76       }
   77   
   78       public void close() {
   79           // guard against multiple closes.
   80           if( closed.compareAndSet(false, true) ) {
   81               worker.addIoTask(new Runnable() {
   82                   public void run() {
   83                       try {
   84                           key.cancel();
   85                       } catch (CancelledKeyException e) {
   86                       }
   87                       worker.release();
   88                   }
   89               });        
   90           }
   91       }
   92   
   93       public void onSelect() {
   94           listener.onSelect(this);
   95       }
   96   
   97       public void onError(Throwable e) {
   98           listener.onError(this, e);
   99       }
  100   
  101   }