Reproduce #983

gistfile1.txt

package org.java_websocket.issues;

import org.java_websocket.WebSocket;
import org.java_websocket.WebSocketImpl;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.framing.CloseFrame;
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.java_websocket.util.ThreadCheck;
import org.junit.*;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assume.assumeThat;

@RunWith(Parameterized.class)
public class Issue983Test {

    private static final int NUMBER_OF_TESTS = 10;
    private static WebSocketServer ws;

    private static int port;
    static CountDownLatch countServerDownLatch = new CountDownLatch( 1 );
    @Rule
    public ThreadCheck zombies = new ThreadCheck();

    @Parameterized.Parameter
    public int count;

    @BeforeClass
    public static void startServer() throws Exception {
        port = SocketUtil.getAvailablePort();
        ws = new WebSocketServer( new InetSocketAddress( port ) , 16) {
            @Override
            public void onOpen( final WebSocket conn, ClientHandshake handshake ) {
                Thread thread = new Thread() {
                    @Override
                    public void run() {
                        for (int i = 0; i < 1000; i++) {
                            if (i % 100 == 0) {
                                conn.sendPing();
                            }
                            conn.send(new byte[10000]);
                        }
                        conn.close();
                    }


                };
                thread.start();
            }

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

            }

            @Override
            public void onMessage( WebSocket conn, String message ) {
                conn.send( message );
            }

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

                ex.printStackTrace(  );
                assumeThat(true, is(false));
                System.out.println("There should be no exception!");
            }

            @Override
            public void onStart() {
                countServerDownLatch.countDown();
            }
        };
        ws.setConnectionLostTimeout( 0 );
        ws.start();
    }

    private void runTestScenario() throws Exception {
        final AtomicInteger counter = new AtomicInteger();
        final AtomicInteger abnormalCloseCounter = new AtomicInteger();
        final AtomicInteger exceptionCounter = new AtomicInteger();
        final CountDownLatch countDownLatchClosed = new CountDownLatch( 1 );

        WebSocketClient clt = new WebSocketClient( new URI( "ws://localhost:" + port ) ) {
            @Override
            public void onOpen( ServerHandshake handshakedata ) {
            }

            @Override
            public void onMessage( String message ) {

            }
            @Override
            public void onMessage(ByteBuffer bytes) {
                counter.incrementAndGet();
            }

            @Override
            public void onClose( int code, String reason, boolean remote ) {
                if (code == CloseFrame.ABNORMAL_CLOSE) {
                    abnormalCloseCounter.incrementAndGet();
                }
                countDownLatchClosed.countDown();
            }

            @Override
            public void onError( Exception ex ) {
                ex.printStackTrace(  );
                exceptionCounter.incrementAndGet();
            }
        };
        clt.connectBlocking();
        countDownLatchClosed.await();
        assertEquals(1000, counter.get());
        assertEquals(0, abnormalCloseCounter.get());
        assertEquals(0, exceptionCounter.get());
    }

    @AfterClass
    public static void successTests() throws InterruptedException, IOException {
        ws.stop();
    }

    @Parameterized.Parameters
    public static Collection<Integer[]> data() {
        List<Integer[]> ret = new ArrayList<Integer[]>(NUMBER_OF_TESTS);
        for (int i = 0; i < NUMBER_OF_TESTS; i++) ret.add(new Integer[]{i});
        return ret;
    }

    @Test(timeout = 5000)
    public void runSendDataWithPingIncluded() throws Exception {
        runTestScenario();
    }
}