@Override
    public void cancel() {
        isRunning = false;

        // we need to close the socket as well, because the Thread.interrupt() function will
        // not wake the thread in the socketStream.read() method when blocked.
        Socket theSocket = this.currentSocket;
        if (theSocket != null) {
            IOUtils.closeSocket(theSocket);
        }
    }
@Test
    void testSocketSourceSimpleOutput() throws Exception {
        ServerSocket server = new ServerSocket(0);
        Socket channel = null;

        try {
            SocketTextStreamFunction source =
                    new SocketTextStreamFunction(LOCALHOST, server.getLocalPort(), "\n", 0);

            SocketSourceThread runner = new SocketSourceThread(source, "test1", "check");
            runner.start();

            channel = NetUtils.acceptWithoutTimeout(server);
            OutputStreamWriter writer = new OutputStreamWriter(channel.getOutputStream());

            writer.write("test1\n");
            writer.write("check\n");
            writer.flush();
            runner.waitForNumElements(2);

            runner.cancel();
            runner.interrupt();

            runner.waitUntilDone();

            channel.close();
        } finally {
            if (channel != null) {
                IOUtils.closeQuietly(channel);
            }
            IOUtils.closeQuietly(server);
        }
    }