Fil d'interruption ne se termine pas appel de blocage sur les flux d'entrée lire

Je suis en utilisant RXTX pour lire des données à partir d'un port série. La lecture se fait dans un thread généré de la manière suivante:

CommPortIdentifier portIdentifier = CommPortIdentifier.getPortIdentifier(port);
CommPort comm = portIdentifier.open("Whatever", 2000);
SerialPort serial = (SerialPort)comm;
...settings
Thread t = new Thread(new SerialReader(serial.getInputStream()));
t.start();

La SerialReader classe implémente Runnable et juste en boucle, la lecture à partir du port et de la construction des données utiles des paquets avant de les envoyer vers d'autres applications. Cependant, je l'ai réduite à la suite de la simplicité:

public void run() {
  ReadableByteChannel byteChan = Channels.newChannel(in); //in = InputStream passed to SerialReader
  ByteBuffer buffer = ByteBuffer.allocate(100);
  while (true) {
    try {
      byteChan.read(buffer);
    } catch (Exception e) {
      System.out.println(e);
    }
  }
}

Lorsqu'un utilisateur clique sur un bouton d'arrêt, les fonctionnalités suivantes feux qui devrait, en théorie, fermer le flux d'entrée et de sortir du blocage byteChan.read(buffer) d'appel. Le code est comme suit:

public void stop() {
  t.interrupt();
  serial.close();
}

Cependant, lorsque j'exécute ce code, je n'obtiens jamais de ClosedByInterruptException, qui DOIT se déclencher une fois que le flux d'entrée se ferme. En outre, l'exécution des blocs sur l'appel à la série.close() -- parce que le sous-jacent flux d'entrée est toujours le blocage de la lecture appel. J'ai essayé de remplacer l'appel d'interruption avec byteChan.close(), qui devrait provoquer une AsynchronousCloseException, cependant, j'obtiens les mêmes résultats.

Toute aide sur ce que je suis absent serait grandement apprécié.

OriginalL'auteur JDS | 2010-10-01

  1. 7

    Vous ne pouvez pas faire un cours d'eau qui ne prend pas en charge interruptible I/O dans un InterruptibleChannel simplement en les enveloppant (et, de toute façon, ReadableByteChannel ne s'étend pas InterruptibleChannel).

    Vous avez à regarder le contrat de sous-jacents InputStream. Ce n' SerialPort.getInputStream() dire à propos de la interruptibility de son résultat? Si elle ne dit rien, vous devez supposer qu'il ignore les interruptions.

    Pour tout I/O qui n'est pas explicitement interruptibility, la seule option est généralement de clôture le flux à partir d'un autre thread. Cela peut immédiatement prendre une IOException (bien qu'il pourrait ne pas être un AsynchronousCloseException) dans le thread bloqué sur un appel à la rivière.

    Cependant, même cela est très dépendante de la mise en œuvre de la InputStream—et de l'OS sous-jacent peut être un facteur aussi.

    Remarque le code source commentaires sur le ReadableByteChannelImpl classe retourné par newChannel():

      private static class ReadableByteChannelImpl
    extends AbstractInterruptibleChannel       //Not really interruptible
    implements ReadableByteChannel
  {
    InputStream in;
    ⋮
    Dans mon exemple, les Canaux.newChannel(<input stream>) retourne un objet de type ReadableByteChannelImpl, qui met en œuvre ReadableByteChannel (mais plus important encore, s'étend AbstractInterruptibleChannel qui implémente InterruptibleChannel).
    Oups...entrez soumet le commentaire. De toute façon, faire un instanceof vérifier byteChan contre InterruptibleChannel renvoie la valeur true. Aussi, puisqu'il n'était pas clair, l'appel à l'arrêt() est fait dans le thread qui engendre la lecture de la boucle de fil.
    et pourtant, ça ne fonctionne pas, n'est ce pas? Veuillez voir ma mise à jour. Le canal n'est pas interruptible, et vous pouvez avoir aucune option viable pour renflouer d'une RXTX lire.

    OriginalL'auteur erickson

  2. 4

    La RXTX SerialInputStream (ce qui est retourné par la série.getInputStream ()) prend en charge un délai d'attente régime qui a fini par résoudre tous mes problèmes. Adjonction, avant la création de la nouvelle SerialReader l'objet provoque le lit pour ne plus bloquer indéfiniment:

    serial.enableReceiveTimeout(1000);

    Dans le SerialReader objet, j'ai dû changer un peu les choses pour les lire directement à partir de l'InputStream au lieu de créer de la ReadableByteChannel, mais maintenant, je peux arrêter et redémarrer le lecteur sans problème.

    Cela fonctionne bien. Merci.

    OriginalL'auteur JDS

  3. 1

    je suis en utilisant le code ci-dessous à l'arrêt rxtx. je exécutent des tests permettant de démarrer et d'arrêter et la semble bien fonctionner. mon lecteur ressemble:

    private void addPartsToQueue(final InputStream inputStream) {
    byte[] buffer = new byte[1024];
    int len = -1;
    boolean first = true;
    //the read can throw
    try {
        while ((len = inputStream.read(buffer)) > -1) {
            if (len > 0) {
                if (first) {
                    first = false;
                    t0 = System.currentTimeMillis();
                } else
                    t1 = System.currentTimeMillis();
                final String part = new String(new String(buffer, 0, len));
                queue.add(part);
                //System.out.println(part + " " + (t1 - t0));
            }
            try {
                Thread.sleep(sleep);
            } catch (InterruptedException e) {
                //System.out.println(Thread.currentThread().getName() + " interrupted " + e);
                break;
            }
        }
    } catch (IOException e) {
        System.err.println(Thread.currentThread().getName() + " " + e);
        //if(interruSystem.err.println(e);
        e.printStackTrace();
    }
    //System.out.println(Thread.currentThread().getName() + " is ending.");
}

    grâce

    public void shutdown(final Device device) {
    shutdown(serialReaderThread);
    shutdown(messageAssemblerThread);
    serialPort.close();
    if (device != null)
        device.setSerialPort(null);
}

public static void shutdown(final Thread thread) {
    if (thread != null) {
        //System.out.println("before intterupt() on thread " + thread.getName() + ", it's state is " + thread.getState());
        thread.interrupt();
        //System.out.println("after intterupt() on thread " + thread.getName() + ", it's state is " + thread.getState());
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            System.out.println(Thread.currentThread().getName() + " was interrupted trying to sleep after interrupting" + thread.getName() + " " + e);
        }
        //System.out.println("before join() on thread " + thread.getName() + ", it's state is " + thread.getState());
        try {
            thread.join();
        } catch (InterruptedException e) {
            System.out.println(Thread.currentThread().getName() + " join interruped");
        }
        //System.out.println(Thread.currentThread().getName() + " after join() on thread " + thread.getName() + ", it's state is" + thread.getState());
    }

    OriginalL'auteur Ray Tayek