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/>.
18 package org.sonews.daemon;
20 import java.io.IOException;
21 import java.net.InetSocketAddress;
22 import java.net.SocketException;
23 import java.nio.ByteBuffer;
24 import java.nio.CharBuffer;
25 import java.nio.channels.ClosedChannelException;
26 import java.nio.channels.SelectionKey;
27 import java.nio.channels.SocketChannel;
28 import java.nio.charset.Charset;
29 import java.util.Arrays;
30 import java.util.Timer;
31 import java.util.TimerTask;
32 import java.util.logging.Level;
33 import org.sonews.daemon.command.Command;
34 import org.sonews.storage.Article;
35 import org.sonews.storage.Group;
36 import org.sonews.storage.StorageBackendException;
37 import org.sonews.util.Log;
38 import org.sonews.util.Stats;
41 * For every SocketChannel (so TCP/IP connection) there is an instance of
43 * @author Christian Lins
46 public final class NNTPConnection {
48 public static final String NEWLINE = "\r\n"; // RFC defines this as newline
49 public static final String MESSAGE_ID_PATTERN = "<[^>]+>";
50 private static final Timer cancelTimer = new Timer(true); // Thread-safe? True for run as daemon
51 /** SocketChannel is generally thread-safe */
52 private SocketChannel channel = null;
53 private Charset charset = Charset.forName("UTF-8");
54 private Command command = null;
55 private Article currentArticle = null;
56 private Group currentGroup = null;
57 private volatile long lastActivity = System.currentTimeMillis();
58 private ChannelLineBuffers lineBuffers = new ChannelLineBuffers();
59 private int readLock = 0;
60 private final Object readLockGate = new Object();
61 private SelectionKey writeSelKey = null;
63 public NNTPConnection(final SocketChannel channel)
65 if (channel == null) {
66 throw new IllegalArgumentException("channel is null");
69 this.channel = channel;
70 Stats.getInstance().clientConnect();
74 * Tries to get the read lock for this NNTPConnection. This method is Thread-
75 * safe and returns true of the read lock was successfully set. If the lock
76 * is still hold by another Thread the method returns false.
78 boolean tryReadLock() {
79 // As synchronizing simple types may cause deadlocks,
80 // we use a gate object.
81 synchronized (readLockGate) {
85 readLock = Thread.currentThread().hashCode();
92 * Releases the read lock in a Thread-safe way.
93 * @throws IllegalMonitorStateException if a Thread not holding the lock
94 * tries to release it.
96 void unlockReadLock() {
97 synchronized (readLockGate) {
98 if (readLock == Thread.currentThread().hashCode()) {
101 throw new IllegalMonitorStateException();
107 * @return Current input buffer of this NNTPConnection instance.
109 public ByteBuffer getInputBuffer() {
110 return this.lineBuffers.getInputBuffer();
114 * @return Output buffer of this NNTPConnection which has at least one byte
117 public ByteBuffer getOutputBuffer() {
118 return this.lineBuffers.getOutputBuffer();
122 * @return ChannelLineBuffers instance associated with this NNTPConnection.
124 public ChannelLineBuffers getBuffers() {
125 return this.lineBuffers;
129 * @return true if this connection comes from a local remote address.
131 public boolean isLocalConnection() {
132 return ((InetSocketAddress) this.channel.socket().getRemoteSocketAddress()).getHostName().equalsIgnoreCase("localhost");
135 void setWriteSelectionKey(SelectionKey selKey) {
136 this.writeSelKey = selKey;
139 public void shutdownInput() {
141 // Closes the input line of the channel's socket, so no new data
142 // will be received and a timeout can be triggered.
143 this.channel.socket().shutdownInput();
144 } catch (IOException ex) {
145 Log.get().warning("Exception in NNTPConnection.shutdownInput(): " + ex);
149 public void shutdownOutput() {
150 cancelTimer.schedule(new TimerTask() {
154 // Closes the output line of the channel's socket.
155 channel.socket().shutdownOutput();
157 } catch (SocketException ex) {
158 // Socket was already disconnected
159 Log.get().info("NNTPConnection.shutdownOutput(): " + ex);
160 } catch (Exception ex) {
161 Log.get().warning("NNTPConnection.shutdownOutput(): " + ex);
167 public SocketChannel getSocketChannel() {
171 public Article getCurrentArticle() {
172 return this.currentArticle;
175 public Charset getCurrentCharset() {
180 * @return The currently selected communication channel (not SocketChannel)
182 public Group getCurrentChannel() {
183 return this.currentGroup;
186 public void setCurrentArticle(final Article article) {
187 this.currentArticle = article;
190 public void setCurrentGroup(final Group group) {
191 this.currentGroup = group;
194 public long getLastActivity() {
195 return this.lastActivity;
199 * Due to the readLockGate there is no need to synchronize this method.
201 * @throws IllegalArgumentException if raw is null.
202 * @throws IllegalStateException if calling thread does not own the readLock.
204 void lineReceived(byte[] raw) {
206 throw new IllegalArgumentException("raw is null");
209 if (readLock == 0 || readLock != Thread.currentThread().hashCode()) {
210 throw new IllegalStateException("readLock not properly set");
213 this.lastActivity = System.currentTimeMillis();
215 String line = new String(raw, this.charset);
217 // There might be a trailing \r, but trim() is a bad idea
218 // as it removes also leading spaces from long header lines.
219 if (line.endsWith("\r")) {
220 line = line.substring(0, line.length() - 1);
221 raw = Arrays.copyOf(raw, raw.length - 1);
224 Log.get().fine("<< " + line);
226 if (command == null) {
227 command = parseCommandLine(line);
228 assert command != null;
232 // The command object will process the line we just received
234 command.processLine(this, line, raw);
235 } catch (StorageBackendException ex) {
236 Log.get().info("Retry command processing after StorageBackendException");
238 // Try it a second time, so that the backend has time to recover
239 command.processLine(this, line, raw);
241 } catch (ClosedChannelException ex0) {
243 StringBuilder strBuf = new StringBuilder();
244 strBuf.append("Connection to ");
245 strBuf.append(channel.socket().getRemoteSocketAddress());
246 strBuf.append(" closed: ");
248 Log.get().info(strBuf.toString());
249 } catch (Exception ex0a) {
250 ex0a.printStackTrace();
252 } catch (Exception ex1) { // This will catch a second StorageBackendException
255 Log.get().log(Level.WARNING, ex1.getLocalizedMessage(), ex1);
256 println("403 Internal server error");
258 // Should we end the connection here?
259 // RFC says we MUST return 400 before closing the connection
262 } catch (Exception ex2) {
263 ex2.printStackTrace();
267 if (command == null || command.hasFinished()) {
269 charset = Charset.forName("UTF-8"); // Reset to default
274 * This method determines the fitting command processing class.
278 private Command parseCommandLine(String line) {
279 String cmdStr = line.split(" ")[0];
280 return CommandSelector.getInstance().get(cmdStr);
284 * Puts the given line into the output buffer, adds a newline character
285 * and returns. The method returns immediately and does not block until
286 * the line was sent. If line is longer than 510 octets it is split up in
287 * several lines. Each line is terminated by \r\n (NNTPConnection.NEWLINE).
290 public void println(final CharSequence line, final Charset charset)
292 writeToChannel(CharBuffer.wrap(line), charset, line);
293 writeToChannel(CharBuffer.wrap(NEWLINE), charset, null);
297 * Writes the given raw lines to the output buffers and finishes with
298 * a newline character (\r\n).
301 public void println(final byte[] rawLines)
303 this.lineBuffers.addOutputBuffer(ByteBuffer.wrap(rawLines));
304 writeToChannel(CharBuffer.wrap(NEWLINE), charset, null);
308 * Encodes the given CharBuffer using the given Charset to a bunch of
309 * ByteBuffers (each 512 bytes large) and enqueues them for writing at the
310 * connected SocketChannel.
311 * @throws java.io.IOException
313 private void writeToChannel(CharBuffer characters, final Charset charset,
314 CharSequence debugLine)
316 if (!charset.canEncode()) {
317 Log.get().severe("FATAL: Charset " + charset + " cannot encode!");
321 // Write characters to output buffers
322 LineEncoder lenc = new LineEncoder(characters, charset);
323 lenc.encode(lineBuffers);
325 enableWriteEvents(debugLine);
328 private void enableWriteEvents(CharSequence debugLine) {
329 // Enable OP_WRITE events so that the buffers are processed
331 this.writeSelKey.interestOps(SelectionKey.OP_WRITE);
332 ChannelWriter.getInstance().getSelector().wakeup();
333 } catch (Exception ex) // CancelledKeyException and ChannelCloseException
335 Log.get().warning("NNTPConnection.writeToChannel(): " + ex);
339 // Update last activity timestamp
340 this.lastActivity = System.currentTimeMillis();
341 if (debugLine != null) {
342 Log.get().fine(">> " + debugLine);
346 public void println(final CharSequence line)
348 println(line, charset);
351 public void print(final String line)
353 writeToChannel(CharBuffer.wrap(line), charset, line);
356 public void setCurrentCharset(final Charset charset) {
357 this.charset = charset;
360 void setLastActivity(long timestamp) {
361 this.lastActivity = timestamp;