Est-il possible de lire à partir d'un InputStream avec un délai d'expiration?

Plus précisément, le problème est d'écrire une méthode comme celle-ci:

int maybeRead(InputStream in, long timeout)

où la valeur de retour est la même que dans in.read () si les données sont disponibles en millisecondes 'timeout', et -2 sinon. Avant le retour de la méthode, tous les threads générés doivent se terminer.

Pour éviter les arguments, le sujet ici java.io.InputStream, tel que documenté par Sun (toute version Java). Veuillez noter que ce n'est pas aussi simple qu'il y paraît. Voici quelques faits qui sont directement corroborés par la documentation de Sun.

1. La méthode in.read () peut être non interruptible.

2. Emballer le InputStream dans un Reader ou InterruptibleChannel n'aide pas, car toutes ces classes peuvent faire est d'appeler des méthodes de InputStream. S'il était possible d'utiliser ces classes, il serait possible d'écrire une solution qui exécute simplement la même logique directement sur InputStream.

3. Il est toujours acceptable que in.available () renvoie 0.

4. La méthode in.close () peut bloquer ou ne rien faire.

5. Il n'existe aucun moyen général de supprimer un autre thread.

Utilisation de inputStream.available ()

Il est toujours acceptable que System.in.available () renvoie 0.

J'ai trouvé le contraire - il renvoie toujours la meilleure valeur pour le nombre d'octets disponibles. Javadoc pour InputStream.available():

Returns an estimate of the number of bytes that can be read (or skipped over) 
from this input stream without blocking by the next invocation of a method for 
this input stream.

Une estimation est inévitable en raison du timing / de l'obsolescence. Le chiffre peut être une sous-estimation ponctuelle car de nouvelles données arrivent constamment. Cependant, il «rattrape» toujours le prochain appel - il doit tenir compte de toutes les données arrivées, sauf celles qui arrivent juste au moment du nouvel appel. Le renvoi permanent de 0 lorsqu'il y a des données échoue à la condition ci-dessus.

Première mise en garde: les sous-classes concrètes d'InputStream sont responsables de available ()

InputStreamest une classe abstraite. Il n'a aucune source de données. Cela n'a aucun sens pour lui d'avoir des données disponibles. Par conséquent, javadoc pour available()indique également:

The available method for class InputStream always returns 0.

This method should be overridden by subclasses.

Et en effet, les classes de flux d'entrée concrètes remplacent available (), fournissant des valeurs significatives, et non des 0 constants.

Deuxième mise en garde: assurez-vous d'utiliser le retour chariot lors de la saisie d'une entrée dans Windows.

Si vous utilisez System.in, votre programme ne reçoit l'entrée que lorsque votre shell de commande le transmet. Si vous utilisez la redirection de fichiers / tubes (par exemple un fichier> java myJavaApp ou une commande | java myJavaApp), les données d'entrée sont généralement transmises immédiatement. Cependant, si vous tapez manuellement l'entrée, le transfert des données peut être retardé. Par exemple, avec le shell Windows cmd.exe, les données sont mises en mémoire tampon dans le shell cmd.exe. Les données ne sont transmises au programme java en cours d'exécution qu'après un retour chariot (control-m ou <enter>). C'est une limitation de l'environnement d'exécution. Bien sûr, InputStream.available () renverra 0 tant que le shell met les données en mémoire tampon - c'est un comportement correct; il n'y a pas de données disponibles à ce stade. Dès que les données sont disponibles depuis le shell, la méthode renvoie une valeur> 0. NB: Cygwin utilise cmd.

Solution la plus simple (pas de blocage, donc pas de timeout requis)

Utilisez simplement ceci:

    byte[] inputData = new byte[1024];
    int result = is.read(inputData, 0, is.available());  
    // result will indicate number of bytes read; -1 for EOF with no data read.

Ou équivalent,

    BufferedReader br = new BufferedReader(new InputStreamReader(System.in, Charset.forName("ISO-8859-1")),1024);
    // ...
         // inside some iteration / processing logic:
         if (br.ready()) {
             int readCount = br.read(inputData, bufferOffset, inputData.length-bufferOffset);
         }

Solution plus riche (remplit au maximum le tampon dans le délai d'expiration)

Déclarez ceci:

public static int readInputStreamWithTimeout(InputStream is, byte[] b, int timeoutMillis)
     throws IOException  {
     int bufferOffset = 0;
     long maxTimeMillis = System.currentTimeMillis() + timeoutMillis;
     while (System.currentTimeMillis() < maxTimeMillis && bufferOffset < b.length) {
         int readLength = java.lang.Math.min(is.available(),b.length-bufferOffset);
         // can alternatively use bufferedReader, guarded by isReady():
         int readResult = is.read(b, bufferOffset, readLength);
         if (readResult == -1) break;
         bufferOffset += readResult;
     }
     return bufferOffset;
 }

Ensuite, utilisez ceci:

    byte[] inputData = new byte[1024];
    int readCount = readInputStreamWithTimeout(System.in, inputData, 6000);  // 6 second timeout
    // readCount will indicate number of bytes read; -1 for EOF with no data read.

En supposant que votre flux ne soit pas soutenu par une socket (vous ne pouvez donc pas l'utiliser Socket.setSoTimeout()), je pense que la manière standard de résoudre ce type de problème est d'utiliser un Future.

Supposons que j'ai l'exécuteur et les flux suivants:

    ExecutorService executor = Executors.newFixedThreadPool(2);
    final PipedOutputStream outputStream = new PipedOutputStream();
    final PipedInputStream inputStream = new PipedInputStream(outputStream);

J'ai un écrivain qui écrit des données puis attend 5 secondes avant d'écrire le dernier élément de données et de fermer le flux:

    Runnable writeTask = new Runnable() {
        @Override
        public void run() {
            try {
                outputStream.write(1);
                outputStream.write(2);
                Thread.sleep(5000);
                outputStream.write(3);
                outputStream.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    };
    executor.submit(writeTask);

La manière normale de lire ceci est la suivante. La lecture bloquera indéfiniment pour les données et cela se termine en 5s:

    long start = currentTimeMillis();
    int readByte = 1;
    // Read data without timeout
    while (readByte >= 0) {
        readByte = inputStream.read();
        if (readByte >= 0)
            System.out.println("Read: " + readByte);
    }
    System.out.println("Complete in " + (currentTimeMillis() - start) + "ms");

qui sort:

Read: 1
Read: 2
Read: 3
Complete in 5001ms

S'il y avait un problème plus fondamental, comme l'écrivain ne répondant pas, le lecteur bloquerait pour toujours. Si j'emballe la lecture dans un futur, je peux alors contrôler le délai comme suit:

    int readByte = 1;
    // Read data with timeout
    Callable<Integer> readTask = new Callable<Integer>() {
        @Override
        public Integer call() throws Exception {
            return inputStream.read();
        }
    };
    while (readByte >= 0) {
        Future<Integer> future = executor.submit(readTask);
        readByte = future.get(1000, TimeUnit.MILLISECONDS);
        if (readByte >= 0)
            System.out.println("Read: " + readByte);
    }

qui sort:

Read: 1
Read: 2
Exception in thread "main" java.util.concurrent.TimeoutException
    at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:228)
    at java.util.concurrent.FutureTask.get(FutureTask.java:91)
    at test.InputStreamWithTimeoutTest.main(InputStreamWithTimeoutTest.java:74)

Je peux attraper le TimeoutException et faire tout le nettoyage que je veux.

Si votre InputStream est sauvegardé par un Socket, vous pouvez définir un délai d'expiration de Socket (en millisecondes) à l'aide de setSoTimeout . Si l'appel read () ne se débloque pas dans le délai spécifié, il lèvera une SocketTimeoutException.

Assurez-vous simplement d'appeler setSoTimeout sur le Socket avant de faire l'appel read ().

Je remettrais en question l'énoncé du problème plutôt que de l'accepter aveuglément. Vous n'avez besoin que de délais d'expiration depuis la console ou via le réseau. Si vous possédez ces derniers Socket.setSoTimeout()et HttpURLConnection.setReadTimeout()que les deux font exactement ce qui est nécessaire, à condition de les configurer correctement lorsque vous les construisez / les acquérez. Le laisser à un point arbitraire plus tard dans l'application lorsque tout ce que vous avez est le InputStream est une mauvaise conception conduisant à une implémentation très délicate.

Je n'ai pas utilisé les classes du package Java NIO, mais il semble qu'elles pourraient être utiles ici. Plus précisément, java.nio.channels.Channels et java.nio.channels.InterruptibleChannel .

Voici un moyen d'obtenir un NIO FileChannel à partir de System.in et de vérifier la disponibilité des données à l'aide d'un délai d'expiration, qui est un cas particulier du problème décrit dans la question. Exécutez-le sur la console, ne tapez aucune entrée et attendez les résultats. Il a été testé avec succès sous Java 6 sur Windows et Linux.

import java.io.FileInputStream;
import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedByInterruptException;

public class Main {

    static final ByteBuffer buf = ByteBuffer.allocate(4096);

    public static void main(String[] args) {

        long timeout = 1000 * 5;

        try {
            InputStream in = extract(System.in);
            if (! (in instanceof FileInputStream))
                throw new RuntimeException(
                        "Could not extract a FileInputStream from STDIN.");

            try {
                int ret = maybeAvailable((FileInputStream)in, timeout);
                System.out.println(
                        Integer.toString(ret) + " bytes were read.");

            } finally {
                in.close();
            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

    /* unravels all layers of FilterInputStream wrappers to get to the
     * core InputStream
     */
    public static InputStream extract(InputStream in)
            throws NoSuchFieldException, IllegalAccessException {

        Field f = FilterInputStream.class.getDeclaredField("in");
        f.setAccessible(true);

        while( in instanceof FilterInputStream )
            in = (InputStream)f.get((FilterInputStream)in);

        return in;
    }

    /* Returns the number of bytes which could be read from the stream,
     * timing out after the specified number of milliseconds.
     * Returns 0 on timeout (because no bytes could be read)
     * and -1 for end of stream.
     */
    public static int maybeAvailable(final FileInputStream in, long timeout)
            throws IOException, InterruptedException {

        final int[] dataReady = {0};
        final IOException[] maybeException = {null};
        final Thread reader = new Thread() {
            public void run() {                
                try {
                    dataReady[0] = in.getChannel().read(buf);
                } catch (ClosedByInterruptException e) {
                    System.err.println("Reader interrupted.");
                } catch (IOException e) {
                    maybeException[0] = e;
                }
            }
        };

        Thread interruptor = new Thread() {
            public void run() {
                reader.interrupt();
            }
        };

        reader.start();
        for(;;) {

            reader.join(timeout);
            if (!reader.isAlive())
                break;

            interruptor.start();
            interruptor.join(1000);
            reader.join(1000);
            if (!reader.isAlive())
                break;

            System.err.println("We're hung");
            System.exit(1);
        }

        if ( maybeException[0] != null )
            throw maybeException[0];

        return dataReady[0];
    }
}

Fait intéressant, lorsque vous exécutez le programme dans NetBeans 6.5 plutôt que sur la console, le délai d'attente ne fonctionne pas du tout et l'appel à System.exit () est en fait nécessaire pour tuer les threads zombies. Ce qui se passe, c'est que le thread d'interruption se bloque (!) Lors de l'appel à reader.interrupt (). Un autre programme de test (non montré ici) essaie également de fermer le canal, mais cela ne fonctionne pas non plus.

Comme jt l'a dit, NIO est la meilleure (et la bonne) solution. Si vous êtes vraiment coincé avec un InputStream, vous pouvez soit

1. Créez un thread dont le travail exclusif consiste à lire à partir de InputStream et à placer le résultat dans un tampon qui peut être lu à partir de votre thread d'origine sans blocage. Cela devrait bien fonctionner si vous n'avez qu'une seule instance du flux. Sinon, vous pourrez peut-être tuer le thread à l'aide des méthodes obsolètes de la classe Thread, bien que cela puisse provoquer des fuites de ressources.

2. Fiez-vous à isAvailable pour indiquer les données qui peuvent être lues sans blocage. Cependant, dans certains cas (comme avec Sockets), il peut prendre une lecture potentiellement bloquante pour isAvailable pour signaler autre chose que 0.

Inspiré par cette réponse, j'ai proposé une solution un peu plus orientée objet.

Ceci n'est valable que si vous avez l'intention de lire des caractères

Vous pouvez remplacer BufferedReader et implémenter quelque chose comme ceci:

public class SafeBufferedReader extends BufferedReader{

    private long millisTimeout;

    ( . . . )

    @Override
    public int read(char[] cbuf, int off, int len) throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return 0;
        }
        return super.read(cbuf, off, len);
    }

    protected void waitReady() throws IllegalThreadStateException, IOException {
        if(ready()) return;
        long timeout = System.currentTimeMillis() + millisTimeout;
        while(System.currentTimeMillis() < timeout) {
            if(ready()) return;
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                break; // Should restore flag
            }
        }
        if(ready()) return; // Just in case.
        throw new IllegalThreadStateException("Read timed out");
    }
}

Voici un exemple presque complet.

Je retourne 0 sur certaines méthodes, vous devriez le changer en -2 pour répondre à vos besoins, mais je pense que 0 est plus approprié avec le contrat BufferedReader. Rien de mal ne s'est produit, il a juste lu 0 caractères. La méthode readLine est un horrible tueur de performances. Vous devez créer un tout nouveau BufferedReader si vous souhaitez réellement utiliser readLin e. À l'heure actuelle, ce n'est pas thread-safe. Si quelqu'un appelle une opération pendant que readLines attend une ligne, cela produira des résultats inattendus

Je n'aime pas retourner -2 où je suis. Je jetterais une exception car certaines personnes peuvent simplement vérifier si int <0 pour considérer EOS. Quoi qu'il en soit, ces méthodes prétendent que "ne peut pas bloquer", vous devriez vérifier si cette déclaration est réellement vraie et ne pas la surcharger.

import java.io.BufferedReader;
import java.io.IOException;
import java.io.Reader;
import java.nio.CharBuffer;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

/**
 * 
 * readLine
 * 
 * @author Dario
 *
 */
public class SafeBufferedReader extends BufferedReader{

    private long millisTimeout;

    private long millisInterval = 100;

    private int lookAheadLine;

    public SafeBufferedReader(Reader in, int sz, long millisTimeout) {
        super(in, sz);
        this.millisTimeout = millisTimeout;
    }

    public SafeBufferedReader(Reader in, long millisTimeout) {
        super(in);
        this.millisTimeout = millisTimeout;
    }



    /**
     * This is probably going to kill readLine performance. You should study BufferedReader and completly override the method.
     * 
     * It should mark the position, then perform its normal operation in a nonblocking way, and if it reaches the timeout then reset position and throw IllegalThreadStateException
     * 
     */
    @Override
    public String readLine() throws IOException {
        try {
            waitReadyLine();
        } catch(IllegalThreadStateException e) {
            //return null; //Null usually means EOS here, so we can't.
            throw e;
        }
        return super.readLine();
    }

    @Override
    public int read() throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return -2; // I'd throw a runtime here, as some people may just be checking if int < 0 to consider EOS
        }
        return super.read();
    }

    @Override
    public int read(char[] cbuf) throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return -2;  // I'd throw a runtime here, as some people may just be checking if int < 0 to consider EOS
        }
        return super.read(cbuf);
    }

    @Override
    public int read(char[] cbuf, int off, int len) throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return 0;
        }
        return super.read(cbuf, off, len);
    }

    @Override
    public int read(CharBuffer target) throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return 0;
        }
        return super.read(target);
    }

    @Override
    public void mark(int readAheadLimit) throws IOException {
        super.mark(readAheadLimit);
    }

    @Override
    public Stream<String> lines() {
        return super.lines();
    }

    @Override
    public void reset() throws IOException {
        super.reset();
    }

    @Override
    public long skip(long n) throws IOException {
        return super.skip(n);
    }

    public long getMillisTimeout() {
        return millisTimeout;
    }

    public void setMillisTimeout(long millisTimeout) {
        this.millisTimeout = millisTimeout;
    }

    public void setTimeout(long timeout, TimeUnit unit) {
        this.millisTimeout = TimeUnit.MILLISECONDS.convert(timeout, unit);
    }

    public long getMillisInterval() {
        return millisInterval;
    }

    public void setMillisInterval(long millisInterval) {
        this.millisInterval = millisInterval;
    }

    public void setInterval(long time, TimeUnit unit) {
        this.millisInterval = TimeUnit.MILLISECONDS.convert(time, unit);
    }

    /**
     * This is actually forcing us to read the buffer twice in order to determine a line is actually ready.
     * 
     * @throws IllegalThreadStateException
     * @throws IOException
     */
    protected void waitReadyLine() throws IllegalThreadStateException, IOException {
        long timeout = System.currentTimeMillis() + millisTimeout;
        waitReady();

        super.mark(lookAheadLine);
        try {
            while(System.currentTimeMillis() < timeout) {
                while(ready()) {
                    int charInt = super.read();
                    if(charInt==-1) return; // EOS reached
                    char character = (char) charInt;
                    if(character == '\n' || character == '\r' ) return;
                }
                try {
                    Thread.sleep(millisInterval);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt(); // Restore flag
                    break;
                }
            }
        } finally {
            super.reset();
        }
        throw new IllegalThreadStateException("readLine timed out");

    }

    protected void waitReady() throws IllegalThreadStateException, IOException {
        if(ready()) return;
        long timeout = System.currentTimeMillis() + millisTimeout;
        while(System.currentTimeMillis() < timeout) {
            if(ready()) return;
            try {
                Thread.sleep(millisInterval);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // Restore flag
                break;
            }
        }
        if(ready()) return; // Just in case.
        throw new IllegalThreadStateException("read timed out");
    }

}