Skip to content

Commit

Permalink
Merge pull request #914 from marci4/Issue900
Browse files Browse the repository at this point in the history
Wrap IOException and include WebSocket
  • Loading branch information
marci4 committed Jan 20, 2020
2 parents ec0fa03 + 72aebe1 commit b52faa4
Show file tree
Hide file tree
Showing 4 changed files with 245 additions and 12 deletions.
2 changes: 1 addition & 1 deletion src/main/java/org/java_websocket/WebSocketImpl.java
Expand Up @@ -510,7 +510,7 @@ public synchronized void closeConnection( int code, String message, boolean remo
try {
channel.close();
} catch ( IOException e ) {
if( e.getMessage().equals( "Broken pipe" ) ) {
if( e.getMessage() != null && e.getMessage().equals( "Broken pipe" ) ) {
log.trace( "Caught IOException: Broken pipe during closeConnection()", e );
} else {
log.error("Exception during channel.close()", e);
Expand Down
@@ -0,0 +1,74 @@
/*
* Copyright (c) 2010-2020 Nathan Rajlich
*
* Permission is hereby granted, free of charge, to any person
* obtaining a copy of this software and associated documentation
* files (the "Software"), to deal in the Software without
* restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the
* Software is furnished to do so, subject to the following
* conditions:
*
* The above copyright notice and this permission notice shall be
* included in all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
* OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
* HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
* OTHER DEALINGS IN THE SOFTWARE.
*
*/

package org.java_websocket.exceptions;

import org.java_websocket.WebSocket;

import java.io.IOException;

/**
* Exception to wrap an IOException and include information about the websocket which had the exception
* @since 1.4.1
*/
public class WrappedIOException extends Exception {

/**
* The websocket where the IOException happened
*/
private final WebSocket connection;

/**
* The IOException
*/
private final IOException ioException;

/**
* Wrapp an IOException and include the websocket
* @param connection the websocket where the IOException happened
* @param ioException the IOException
*/
public WrappedIOException(WebSocket connection, IOException ioException) {
this.connection = connection;
this.ioException = ioException;
}

/**
* The websocket where the IOException happened
* @return the websocket for the wrapped IOException
*/
public WebSocket getConnection() {
return connection;
}

/**
* The wrapped IOException
* @return IOException which is wrapped
*/
public IOException getIOException() {
return ioException;
}
}
28 changes: 17 additions & 11 deletions src/main/java/org/java_websocket/server/WebSocketServer.java
Expand Up @@ -47,6 +47,7 @@
import org.java_websocket.*;
import org.java_websocket.drafts.Draft;
import org.java_websocket.exceptions.WebsocketNotConnectedException;
import org.java_websocket.exceptions.WrappedIOException;
import org.java_websocket.framing.CloseFrame;
import org.java_websocket.framing.Framedata;
import org.java_websocket.handshake.ClientHandshake;
Expand Down Expand Up @@ -320,7 +321,6 @@ public void run() {
int selectTimeout = 0;
while ( !selectorthread.isInterrupted() && iShutdownCount != 0) {
SelectionKey key = null;
WebSocketImpl conn = null;
try {
if (isclosed.get()) {
selectTimeout = 5;
Expand All @@ -334,7 +334,6 @@ public void run() {

while ( i.hasNext() ) {
key = i.next();
conn = null;

if( !key.isValid() ) {
continue;
Expand All @@ -358,10 +357,10 @@ public void run() {
// an other thread may cancel the key
} catch ( ClosedByInterruptException e ) {
return; // do the same stuff as when InterruptedException is thrown
} catch ( WrappedIOException ex) {
handleIOException( key, ex.getConnection(), ex.getIOException());
} catch ( IOException ex ) {
if( key != null )
key.cancel();
handleIOException( key, conn, ex );
handleIOException( key, null, ex );
} catch ( InterruptedException e ) {
// FIXME controlled shutdown (e.g. take care of buffermanagement)
Thread.currentThread().interrupt();
Expand Down Expand Up @@ -445,7 +444,7 @@ private void doAccept(SelectionKey key, Iterator<SelectionKey> i) throws IOExcep
* @throws InterruptedException thrown by taking a buffer
* @throws IOException if an error happened during read
*/
private boolean doRead(SelectionKey key, Iterator<SelectionKey> i) throws InterruptedException, IOException {
private boolean doRead(SelectionKey key, Iterator<SelectionKey> i) throws InterruptedException, WrappedIOException {
WebSocketImpl conn = (WebSocketImpl) key.attachment();
ByteBuffer buf = takeBuffer();
if(conn.getChannel() == null){
Expand All @@ -471,7 +470,7 @@ private boolean doRead(SelectionKey key, Iterator<SelectionKey> i) throws Interr
}
} catch ( IOException e ) {
pushBuffer( buf );
throw e;
throw new WrappedIOException(conn, e);
}
return true;
}
Expand All @@ -481,12 +480,16 @@ private boolean doRead(SelectionKey key, Iterator<SelectionKey> i) throws Interr
* @param key the selectionkey to write on
* @throws IOException if an error happened during batch
*/
private void doWrite(SelectionKey key) throws IOException {
private void doWrite(SelectionKey key) throws WrappedIOException {
WebSocketImpl conn = (WebSocketImpl) key.attachment();
if( SocketChannelIOHelper.batch( conn, conn.getChannel() ) ) {
if( key.isValid() ) {
key.interestOps(SelectionKey.OP_READ);
try {
if (SocketChannelIOHelper.batch(conn, conn.getChannel())) {
if (key.isValid()) {
key.interestOps(SelectionKey.OP_READ);
}
}
} catch (IOException e) {
throw new WrappedIOException(conn, e);
}
}

Expand Down Expand Up @@ -598,6 +601,9 @@ private void pushBuffer( ByteBuffer buf ) throws InterruptedException {

private void handleIOException( SelectionKey key, WebSocket conn, IOException ex ) {
// onWebsocketError( conn, ex );// conn may be null here
if (key != null) {
key.cancel();
}
if( conn != null ) {
conn.closeConnection( CloseFrame.ABNORMAL_CLOSE, ex.getMessage() );
} else if( key != null ) {
Expand Down
153 changes: 153 additions & 0 deletions src/test/java/org/java_websocket/issues/Issue900Test.java
@@ -0,0 +1,153 @@
/*
* Copyright (c) 2010-2020 Nathan Rajlich
*
* Permission is hereby granted, free of charge, to any person
* obtaining a copy of this software and associated documentation
* files (the "Software"), to deal in the Software without
* restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the
* Software is furnished to do so, subject to the following
* conditions:
*
* The above copyright notice and this permission notice shall be
* included in all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
* OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
* HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
* OTHER DEALINGS IN THE SOFTWARE.
*
*/

package org.java_websocket.issues;

import org.java_websocket.WebSocket;
import org.java_websocket.WebSocketImpl;
import org.java_websocket.WrappedByteChannel;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ClientHandshake;
import org.java_websocket.handshake.ServerHandshake;
import org.java_websocket.server.WebSocketServer;
import org.java_websocket.util.SocketUtil;
import org.junit.Test;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.concurrent.CountDownLatch;

public class Issue900Test {

CountDownLatch serverStartLatch = new CountDownLatch(1);
CountDownLatch closeCalledLatch = new CountDownLatch(1);

@Test(timeout = 2000)
public void testIssue() throws Exception {
int port = SocketUtil.getAvailablePort();
final WebSocketClient client = new WebSocketClient(new URI("ws://localhost:" + port)) {
@Override
public void onOpen(ServerHandshake handshakedata) {

}

@Override
public void onMessage(String message) {
}

@Override
public void onClose(int code, String reason, boolean remote) {
}

@Override
public void onError(Exception ex) {

}
};
WebSocketServer server = new WebSocketServer(new InetSocketAddress(port)) {
@Override
public void onOpen(WebSocket conn, ClientHandshake handshake) {
}

@Override
public void onClose(WebSocket conn, int code, String reason, boolean remote) {
closeCalledLatch.countDown();
}

@Override
public void onMessage(WebSocket conn, String message) {

}

@Override
public void onError(WebSocket conn, Exception ex) {

}

@Override
public void onStart() {
serverStartLatch.countDown();
}
};
new Thread(server).start();
serverStartLatch.await();
client.connectBlocking();
WebSocketImpl websocketImpl = (WebSocketImpl)new ArrayList<WebSocket>(server.getConnections()).get(0);
websocketImpl.setChannel(new ExceptionThrowingByteChannel());
server.broadcast("test");
closeCalledLatch.await();
}
class ExceptionThrowingByteChannel implements WrappedByteChannel {

@Override
public boolean isNeedWrite() {
return true;
}

@Override
public void writeMore() throws IOException {
throw new IOException();
}

@Override
public boolean isNeedRead() {
return true;
}

@Override
public int readMore(ByteBuffer dst) throws IOException {
throw new IOException();
}

@Override
public boolean isBlocking() {
return false;
}

@Override
public int read(ByteBuffer dst) throws IOException {
throw new IOException();
}

@Override
public int write(ByteBuffer src) throws IOException {
throw new IOException();
}

@Override
public boolean isOpen() {
return false;
}

@Override
public void close() throws IOException {
throw new IOException();
}
}
}

0 comments on commit b52faa4

Please sign in to comment.