InputStream not detecting when in the other Side the OuputStrem and Socket are Closed

Question

I made an example of my situation:

I have a Client producing Doubles Numbers (Really I need send bytes) in the other side I have a Server attending the clients request (for this example only One), and creates an Worker...

The problem is that the Worker (with InputStream created according to Socket accepted by Server) is not detecting when the Client (closing the OutputStream and Socket) is terminated.

Here my code:

Client

class ProducerDouble extends Thread {
  private Socket consSocket = null;
  private OutputStream output = null;
  private byte[] innerBytes = null;
  private boolean isRunning = false;
  public ProducerDouble(String consHost , int consPort) {
    try {
      consSocket = new Socket(consHost, consPort);
    }
    catch (UnknownHostException e) {
      System.out.println("Unknown host!");
    }
    catch (IOException e) {
      System.out.println("Consumer :"+e.getMessage());
    }
    try {
      if (!(consSocket == null)) {
        output = consSocket.getOutputStream();
      }
    }
    catch (IOException e) { }
  }
  @Override
  public void run() {
    if (!(consSocket == null)) {
      isRunning = true;
      int read = 0;
      while (isRunning) {
        try {
          if (!(innerBytes == null)) {
            output.write(innerBytes, 0, innerBytes.length);
            System.out.println("ProducerDouble "+innerBytes.length +" bytes written");
            innerBytes = null;
          } else {
            try {
              Thread.sleep(2000);
              String sSending = "The generated double is:"+Double.toString(Math.random()*100.0);
              innerBytes = sSending.getBytes();
            } catch (InterruptedException ex) { }
          }
        } catch (IOException e) { }
      }
      try {
        output.close();
      } catch (IOException e) { }
      try {
        consSocket.close();
      } catch (IOException e) { }
      System.out.println("\n Exiting ProducerDouble...\n");
    }
  }

  public void stopExecute() {
    isRunning = false;
  }
}

Worker

class ConsumerDouble extends Thread {
  private Socket clientSocket = null;
  private InputStream input = null;
  private byte[] innerBytes = null;
  private boolean isRunning = false;

  public ConsumerDouble(Socket newClientSocket) {
    clientSocket = newClientSocket;
    try {
      input  = clientSocket.getInputStream();
    } catch (IOException e) { }
  }
  @Override
  public void run() {
    isRunning = true;
    while (isRunning) {
      System.out.println("ConsumerDouble running");
      try {
        if (innerBytes == null) {
          if (input.available() > 0) {
            innerBytes = new byte[input.available()];
            int read = input.read(innerBytes, 0, innerBytes.length);
            System.out.println("ConsumerDouble " + innerBytes.length 
              +" bytes read from Host ");
          } else {
            try {
              Thread.sleep(1500);
            } catch (InterruptedException ex) { }
          }
        }
      } catch (IOException e) {
        isRunning = false;
        System.out.println("\nConsumerDouble IOException:"+e.getMessage()+"\n");
      }
    }
    System.out.println("\n ConsumerDouble Request Terminated...\n");
    try {  if (input != null) input.close(); } catch (IOException e) { }
    try {  if (clientSocket != null) clientSocket.close(); } catch (IOException e) { }
  }

  public void stopExecute() {
    isRunning = false;
  }
}

Launching and Stopping the Server

if (jToggleButtonServer.isSelected()) {
  thrdSrvrDouble = new Thread() {
    @Override
    public void run() {
      try {
        SrvrSocketDouble = new ServerSocket(1023);
        System.out.println("ServerDouble Listening on port number: "+1023);
      } catch (IOException e) {
        System.out.println("Could not listen on port: "+1023);
      }
      bThrdServerDoubleRunning = true;
      while (bThrdServerDoubleRunning) {
        Socket clientSocket = null;
        try {
          clientSocket = SrvrSocketDouble.accept();
          System.out.println("New Client Address: " + clientSocket.getInetAddress() + "  Port:" + clientSocket.getPort());
        } catch (IOException e) {
          if(!bThrdServerDoubleRunning) {
            System.out.println("ServerDouble Stopped.") ;
            break;
          }
          throw new RuntimeException("Error: ServerDouble accepting client connections", e);
        }
        if (consumerDouble == null) {
          consumerDouble = new ConsumerDouble(clientSocket);
          consumerDouble.start();
        }
      }
      consumerDouble.stopExecute();
      System.out.println("\n  Exiting ServerDouble!!!");
    }
  };
  thrdSrvrDouble.start();
} else  {
  bThrdServerDoubleRunning = false;
  try {
    SrvrSocketDouble.close();
  } catch (IOException e) { }
}

Variables

static ServerSocket SrvrSocketDouble;
Thread thrdSrvrDouble = null;
static boolean bThrdServerDoubleRunning = false;
ProducerDouble producerDouble = null;

Launching and Stopping the Client (ProducerDouble)

if (jToggleButtonProducer.isSelected()) {
  String sConsHost = jtfSrClConsHost.getText();
  producerDouble = new ProducerDouble(sConsHost, 1023);
  producerDouble.start();
} else {
  producerDouble.stopExecute();
}

The OUPUT

run:
ServerDouble Listening on port number: 1023
New Client Address: /192.168.0.16  Port:51056
ConsumerDouble running
ConsumerDouble running
ProducerDouble 41 bytes written
ConsumerDouble running
ConsumerDouble 41 bytes read from Host 
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ProducerDouble 41 bytes written

 Exiting ProducerDouble...

ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ConsumerDouble running
ServerDouble Stopped.

  Exiting ServerDouble!!!

 ConsumerDouble Request Terminated...

BUILD SUCCESSFUL (total time: 2 minutes 0 seconds)

When ProducerDouble finishes Exiting ProducerDouble... at least the ConsumerDouble must be finalized too printing the Message ConsumerDouble IOException: (the message Exception)!

Question

1. What is the reason for the exception is not thrown into the while Loop of Worker (ConsumerDouble)?
2. How Can I solve this?

PD: I was reading...

How to detect a remote side socket close? but for me OutputStream is working detecting when InputStream is closed (the problem is with InputStream is not detecting when OutputStream is closed even if the Socket creator of output (in the other side) is closed)!

Socket close vs Inputstream close is not related to my question.

Thank you!

Answer 1

1. What is the reason for the exception is not thrown into the while Loop of Worker (ConsumerDouble)?

Why should it be? What exception? What are you expecting? The way that InputStream.read() signals end of stream is by returning -1. However as you are never testing or even displaying that value, you are never detecting end of stream. The problem is made worse by your misuse of available() which you should just remove. At present you are only calling read() when there is data to be read immediately, which will never be true at end of stream.

2. How Can I solve this?

Test for read() returning -1.

You also have a misleading message where you claim that the length of the buffer is the number of bytes that were just read. It isn't. You are also ignoring IOException in the client. Don't do that.

Answer 2

You need change something in your Worker to detect when on the other side it has closed the OutputStream or Socket (InputStream counterpart to be verified), throwing Exceptions.

FIRST OPTION

Taking advantage that you do not use OutputStream (in ConsumerDouble) could force an exception sending bytes.

isRunning = true;
while (isRunning) {
  try {
    clientSocket.getOutputStream().write(0);
  } catch (IOException e) {
    isRunning = false;
    //  Software caused connection abort: socket write error
  }
  System.out.println("ConsumerDouble running");
  // from here on the rest of the code will remain the same as it is in your.

SECOND OPTION

You need to change the receive logic used

class ConsumerDouble extends Thread {
  private Socket clientSocket = null;
  private InputStream input = null;
  private byte[] innerBytes = new byte [1024];  // Is not null!
  private boolean isRunning = false;
  private int read = 0;  // moved from while loop!

Completely changing the While loop!!

    while (isRunning) {
      System.out.println("ConsumerDouble running");
      try {
        read = input.read(innerBytes, 0, innerBytes.length);
        System.out.println("ConsumerDouble " + read 
          +" bytes read from Host ");
        byte[] nBytes = new byte[read];
        System.arraycopy(innerBytes, 0, nBytes, 0, read);
        System.out.println("Received: " + new String(nBytes));
      } catch (IOException e) {
        isRunning = false;
         System.out.println("\nConsumerDouble IOException:"+e.getMessage()+"\n");
      } catch (NegativeArraySizeException e) {
        isRunning = false;
         System.out.println("\nConsumerDouble NegativeArraySizeException:"+e.getMessage()+"\n");
      }
    }

Following the recommendation...

    while (isRunning) {
      System.out.println("ConsumerDouble running");
      try {
        read = input.read(innerBytes, 0, innerBytes.length);
        if (read != -1) {
          System.out.println("ConsumerDouble " + read 
              +" bytes read from Host ");
          // this code is an example showing the message
          byte[] nBytes = new byte[read];
          System.arraycopy(innerBytes, 0, nBytes, 0, read);
          System.out.println("Received: " + new String(nBytes));
        } else {
          isRunning = false;  // the Stream was closed!
        }
      } catch (IOException e) {
        isRunning = false;
         System.out.println("\nConsumerDouble IOException:"+e.getMessage()+"\n");
      }
    }