PullFeeder sends an addition "MODE READER" to peers.
3 * see AUTHORS for the list of contributors
5 * This program is free software: you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License as published by
7 * the Free Software Foundation, either version 3 of the License, or
8 * (at your option) any later version.
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
15 * You should have received a copy of the GNU General Public License
16 * along with this program. If not, see <http://www.gnu.org/licenses/>.
19 package org.sonews.daemon;
21 import java.io.IOException;
22 import java.nio.ByteBuffer;
23 import java.nio.channels.CancelledKeyException;
24 import java.nio.channels.SelectionKey;
25 import java.nio.channels.Selector;
26 import java.nio.channels.SocketChannel;
27 import java.util.Iterator;
29 import org.sonews.util.Log;
32 * A Thread task listening for OP_READ events from SocketChannels.
33 * @author Christian Lins
36 class ChannelReader extends AbstractDaemon
39 private static ChannelReader instance = new ChannelReader();
42 * @return Active ChannelReader instance.
44 public static ChannelReader getInstance()
49 private Selector selector = null;
51 protected ChannelReader()
56 * Sets the selector which is used by this reader to determine the channel
60 public void setSelector(final Selector selector)
62 this.selector = selector;
66 * Run loop. Blocks until some data is available in a channel.
71 assert selector != null;
77 // select() blocks until some SelectableChannels are ready for
78 // processing. There is no need to lock the selector as we have only
79 // one thread per selector.
82 // Get list of selection keys with pending events.
83 // Note: the selected key set is not thread-safe
84 SocketChannel channel = null;
85 NNTPConnection conn = null;
86 final Set<SelectionKey> selKeys = selector.selectedKeys();
87 SelectionKey selKey = null;
89 synchronized (selKeys)
91 Iterator it = selKeys.iterator();
93 // Process the first pending event
96 selKey = (SelectionKey) it.next();
97 channel = (SocketChannel) selKey.channel();
98 conn = Connections.getInstance().get(channel);
100 // Because we cannot lock the selKey as that would cause a deadlock
101 // we lock the connection. To preserve the order of the received
102 // byte blocks a selection key for a connection that has pending
103 // read events is skipped.
104 if (conn == null || conn.tryReadLock())
106 // Remove from set to indicate that it's being processed
110 break; // End while loop
122 // Do not lock the selKeys while processing because this causes
123 // a deadlock in sun.nio.ch.SelectorImpl.lockAndDoSelect()
124 if (selKey != null && channel != null && conn != null)
126 processSelectionKey(conn, channel, selKey);
127 conn.unlockReadLock();
131 catch(CancelledKeyException ex)
133 Log.msg("ChannelReader.run(): " + ex, false);
136 ex.printStackTrace();
141 ex.printStackTrace();
144 // Eventually wait for a register operation
145 synchronized (NNTPDaemon.RegisterGate)
147 // Do nothing; FindBugs may warn about an empty synchronized
148 // statement, but we cannot use a wait()/notify() mechanism here.
149 // If we used something like RegisterGate.wait() we block here
150 // until the NNTPDaemon calls notify(). But the daemon only
151 // calls notify() if itself is NOT blocked in the listening socket.
153 } // while(isRunning())
156 private void processSelectionKey(final NNTPConnection connection,
157 final SocketChannel socketChannel, final SelectionKey selKey)
158 throws InterruptedException, IOException
160 assert selKey != null;
161 assert selKey.isReadable();
163 // Some bytes are available for reading
167 //synchronized(socketChannel)
169 // Read the data into the appropriate buffer
170 ByteBuffer buf = connection.getInputBuffer();
174 read = socketChannel.read(buf);
176 catch(IOException ex)
178 // The connection was probably closed by the remote host
179 // in a non-clean fashion
180 Log.msg("ChannelReader.processSelectionKey(): " + ex, true);
184 Log.msg("ChannelReader.processSelectionKey(): " + ex, false);
187 ex.printStackTrace();
191 if(read == -1) // End of stream
195 else if(read > 0) // If some data was read
197 ConnectionWorker.addChannel(socketChannel);
204 Log.msg(selKey, false);