Привет всем.
Похоже, придется мне скоро нырнуть в джаву без скафандра, так что скоро заспамлю этот форум вопросами.
Это первый.

Я делаю тренировочный проект — клиентскую программку, которая общается с сервером по сети через потоки, которые я снимаю с сокета.
Т.е. есть объект, реализующий функционал, ему при создании передаются эти потоки.

Однако, в целях юнит-тестирования я хочу использовать не потоки с сокета, а потоки,
которые могу наполнять сам — например, ByteArrayInputStream/ByteArrayOutputStream.
Т.е. создаю входной и выходной потоки, и объект тестируемого класса, которому эти потоки отдаются.

Хочу протестировать, что при отправки во входной поток определенного сообщения на выходе будет ожидаемое сообщение.

Проблема в том, что входные данные внутри тестируемого объекта могут перевариваться некоторое время (зависит от теста),
и данные в выходной поток поступят не сразу.

Вопрос: как правильно их дождаться?

Можно просто подождать заведомо достаточное время, но это как-то не круто, привносит лишнюю задержку.
Можно в цикле дергать toByteArray() и смотреть, нет ли в получившемся буфере маркера конца сообщения.
Но это какой-то совсем некрасивый подход, да еще cpu жрет.

В идеале, хотелось бы какой-то аналог плюсового basic_iostream, чтобы тестируемый класс туда писал, а тест оттуда читал и проверял данные.

Как в джаве принято такое делать? Писать свой класс потока?

Здравствуйте, f95.2, Вы писали:

F2>Как в джаве принято такое делать? Писать свой класс потока?

Есть не пугает работа с Threads, то можно изпользовать PipedReader/PipedWriter (или PipedInputStream/PipedOutputStream). Они связываются через метод connect() любого из объектов и становятся чем-то вроде Unix pipes, то есть читатель автоматически блокируется, если данных нет. С помощью адаптеров из Channels можно оборачивать ReadableByteChannel и WritableByteChannel.

Здравствуйте, f95.2, Вы писали:

F2>Можно просто подождать заведомо достаточное время, но это как-то не круто, привносит лишнюю задержку.
Так точно не надо.

F2>Можно в цикле дергать toByteArray() и смотреть, нет ли в получившемся буфере маркера конца сообщения.
F2>Но это какой-то совсем некрасивый подход, да еще cpu жрет.
Ждать c таймаутом правильнее.
Ну как бы CPU не самое критичное в юнит тестах, и sleep решает эту проблему.
Есть либы для помощи в таких делах https://github.com/awaitility/awaitility

@Test
public void updatesCustomerStatus() {
    // Publish an asynchronous message to a broker (e.g. RabbitMQ):
    messageBroker.publishMessage(updateCustomerStatusMessage);
    // Awaitility lets you wait until the asynchronous operation completes:
    await().atMost(5, SECONDS).until(customerStatusIsUpdated());
    ...
}

Еще как вариант параметризовать тестируемый класс, чтоб вся обработка синхронно происходила. Для проверки логики, не связанной с асинхронным выполнением, самый правильный вариант.

F2>В идеале, хотелось бы какой-то аналог плюсового basic_iostream, чтобы тестируемый класс туда писал, а тест оттуда читал и проверял данные.
Для тестов ByteArrayInputStream и ByteArrayOutputStream подходят более чем.

Можно не через стримы общаться а через свой интерфейс. В боевоей конфигурации StreamPublisher в тестах TestPublisher который удобен для проверки результата.

Здравствуйте, f95.2, Вы писали:

f> Хочу протестировать, что при отправки во входной поток определенного сообщения на выходе будет ожидаемое сообщение.
Я бы разделил код на части — одна часть преобразует потоки в сообщения, т.е. занимается транспортом, а другая обрабатывает сообщения, т.е. бизнес-логика. Тогда бизнес-логика будет тестироваться элементарно. И отдельно покрыть функционал преобразования потоков в сообщения. А ещё лучше взять какую-нибудь готовую библиотечку для транспорта, их просто дохрена.

f> Вопрос: как правильно их дождаться?
Именно в юнит-тестах очень желательно избегать многопоточность, паузы и прочее.
А в интеграционных тестах желательно таки всё сделать как можно ближе к реальности — реальные сокеты и т.п.

С этими классами напоролся на неожиданную проблему:
Допустим, поток читатель ждет (т.е. использует блокирующее чтение) данных в потоке.
В это время другой поток определил, что пользователь хочет завершить программу.

Нужно как-то вывести поток-читатель из заблокированного состояния.
Как?
Thread.interrupt? Это некрасиво, да и вроде бы interrupt считается устаревшим.
Я пытался закрыть InputStream с расчетом на то, что блокирующее чтение выбросит исключение, и поток сможет определить,
что читать дальше не надо.

Но исключение почему-то не выбласывается.
Вот пример программы, на которой воспроизводится подобное поведение:

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;

public class Main {
    public static void main(String[] args) throws IOException, InterruptedException {
        PipedInputStream in = new PipedInputStream();
        PipedOutputStream pipedHelper = new PipedOutputStream(in);

        Thread th = new Thread(() -> {
            try {
                in.read();
            } catch (IOException e) {
                System.out.println("Exception in thread");
            }
        });

        th.start();

        Thread.sleep(1000);

        in.close();

        th.join();

        System.out.println("Done");
    }
}

Основной поток ждет на th.join(), а дополнительный — на in.read().

Как правильно разбудить читающий поток?

Здравствуйте, ·, Вы писали:

·>Я бы разделил код на части — одна часть преобразует потоки в сообщения
Так я так и сделал

·>А ещё лучше взять какую-нибудь готовую библиотечку для транспорта, их просто дохрена.
У меня сейчас цель не сделать программу на продажу, а изучить язык и его стандартную библиотеку.
Поэтому готовая библиотека мне не подходит.

·>Именно в юнит-тестах очень желательно избегать многопоточность, паузы и прочее.
Это еще почему?

·>А в интеграционных тестах желательно таки всё сделать как можно ближе к реальности — реальные сокеты и т.п.
Во-первых, я не делю тесты только на интеграционные и юнит. У меня есть несколько промежуточных стадий
А во-вторых, арендовать сервер под учебный пет-проект — это перебор, я считаю.

Здравствуйте, f95.2, Вы писали:

F2>·>Я бы разделил код на части — одна часть преобразует потоки в сообщения
F2>Так я так и сделал


F2>·>Именно в юнит-тестах очень желательно избегать многопоточность, паузы и прочее.
F2>Это еще почему?
Потому что твои юнит-тесты должны тестировать твой код. Тестировать как работают потоки, сокеты и прочее — это задача разработчиков операционки и jdk.
Такие юнит-тесты очень быстры, их легко конфигурировать и отлаживать.

А интеграционные тесты должны тестировать как твой код интегрируется с кодом операционки и jdk.

F2>·>А в интеграционных тестах желательно таки всё сделать как можно ближе к реальности — реальные сокеты и т.п.
F2>Во-первых, я не делю тесты только на интеграционные и юнит. У меня есть несколько промежуточных стадий
Месиво т.е.?

F2>А во-вторых, арендовать сервер под учебный пет-проект — это перебор, я считаю.
Какой ещё сервер? Сокеты могут через localhost работать.

·>Потому что твои юнит-тесты должны тестировать твой код. Тестировать как работают потоки, сокеты и прочее — это задача разработчиков операционки и jdk.
В плюсах я привык полагаться на то, что стандартная библиотека работает без ошибок.
Естественно, баги могут быть везде, но вероятность их появления в реализации stl от компилятора все же очень мала,
и обходить их костылями можно только после явного обнаружения.

В мире java не так? Там не принято доверять реализации стандартной библиотеки?
Или почему ты заговорил про тестирование jdk?


·>Такие юнит-тесты очень быстры, их легко конфигурировать и отлаживать.
·>А интеграционные тесты должны тестировать как твой код интегрируется с кодом операционки и jdk.

Я просто по-другому понимаю названия типов тестов.
У меня юнит-тесты — это тесты, которые проверяют обособленные (и, обычно, мелкие) куски функционала.
Например, парсер чего-нибудь.
А интеграционные тесты — это тесты, которые проверяют взаимосвязь нескольких компонентов системы вместе.

Если система представляет собой многоуровневую абстракцию (крупные компоненты дробятся на более мелкие, те на еще более мелкие и т. д.),
то на каждый такой уровень свой набор тестов. Поэтому я и написал про несколько уровней.

Но это оффтоп.

·>Какой ещё сервер? Сокеты могут через localhost работать.
так localhost — это недостаточно близко к реальности :D

Здравствуйте, f95.2, Вы писали:

F2>·>Потому что твои юнит-тесты должны тестировать твой код. Тестировать как работают потоки, сокеты и прочее — это задача разработчиков операционки и jdk.
F2>В плюсах я привык полагаться на то, что стандартная библиотека работает без ошибок.
F2>Естественно, баги могут быть везде, но вероятность их появления в реализации stl от компилятора все же очень мала,
F2>и обходить их костылями можно только после явного обнаружения.
F2>В мире java не так? Там не принято доверять реализации стандартной библиотеки?
F2>Или почему ты заговорил про тестирование jdk?
Так я это имею в виду — в юнит-тестах тестировать jdk не надо. Т.е. кода из jdk, а тем более треды, сокеты, етс — в юнит-тестах должно быть по минимуму.
Тестировать саму jdk вообще не надо. Надо тестировать как твой код интегрируется с jdk.

F2>·>Такие юнит-тесты очень быстры, их легко конфигурировать и отлаживать.
F2>·>А интеграционные тесты должны тестировать как твой код интегрируется с кодом операционки и jdk.
F2>Я просто по-другому понимаю названия типов тестов.
F2>У меня юнит-тесты — это тесты, которые проверяют обособленные (и, обычно, мелкие) куски функционала.
F2>Например, парсер чего-нибудь.
F2>А интеграционные тесты — это тесты, которые проверяют взаимосвязь нескольких компонентов системы вместе.
Понятно... В принципе да, тут вопрос терминологии. Я это просто не в абстрактных компонентах рассуждаю, а с т.з. того чего там происходит. И юнит-тесты ещё можно понимать что они тестируют собственно сам код, а не внешние к языку вещи типа сокетов, потоков, етс.

F2>Если система представляет собой многоуровневую абстракцию (крупные компоненты дробятся на более мелкие, те на еще более мелкие и т. д.),
F2>то на каждый такой уровень свой набор тестов. Поэтому я и написал про несколько уровней.
Юнит — это модуль, т.е. это не про размер а про варианты использования. Т.е. если у тебя есть реюзабельный модуль, который можно куда-то втыкать пусть большой, то можно и потестировать юнит-тестами. Т.е. если у тебя особым образом есть сконфигурированный планигами парсер — то его можно тестировать как единый модуль.

F2>Но это оффтоп.
Ага. И вопрос терминологии.

F2>·>Какой ещё сервер? Сокеты могут через localhost работать.
F2>так localhost — это недостаточно близко к реальности :D
Эээ... Зависит от. Полагаю, для типичного учебного пет-проекта — более чем достаточно близко.

Здравствуйте, f95.2, Вы писали:

f>         Thread.sleep(1000);
f>         in.close();

Тут надо закрывать писателя pipedHelper.close()

·>Тут надо закрывать писателя pipedHelper.close()

Так а что делать, если известен только InputStream?
Еще раз, есть вот такой объект:

// Сообщения специального формата. Парсятся из входного потока, ответы на них пишутся в выходной поток
class Message {
    ....
}


class Proxy {
    public Proxy(InputStream in, OutputStream out, Consumer<Message> cons) {
        ....
    }

    public void stop() {
        ....
    }
}

При создании объект Proxy внутри себя запускает вспомогательные потоки. В них он ждет данных из входного потока,
пишет данные в выходной поток, периодически вызывает пользовательский обработчик, аргумент которого — распарсенное сообщение.

Предполагается, что в реальной жизни потоки будут взяты с сокета, а в тестах — созданы руками.
Но сам класс не знает, где он запускается, и видит только InputStream и OutputStream.

В один прекрасный момент из главного потока (так проще, про вызов из обработчика я еще подумаю) вызывается Proxy.stop(),
который ждет, пока Proxy остановит все свои потоки и освободит все занятые ресурсы.

Как это сделать, если некоторые потоки заблокированы на чтении данных?
Я поначалу хотел просто закрыть входящий поток, но, похоже, этого недостаточно.

Здравствуйте, f95.2, Вы писали:

f> В один прекрасный момент из главного потока (так проще, про вызов из обработчика я еще подумаю) вызывается Proxy.stop(),
f> который ждет, пока Proxy остановит все свои потоки и освободит все занятые ресурсы.
Вот это непонятно. Очевидно, так нельзя делать. В реальной ситуации у тебя закроется источник, т.е. сам сокет, и это прервёт read в InputStream. Значит и в тесте надо закрывать источник, т.е. этот твой pipedHelper. Закрывать сам стрим из другого потока как-то странно, имхо, он вовсе не обязан быть interruptible, т.е. read может сидеть в глубинах сискола и треды, тем более на уровне языка, тут ничего сделать не могут.
https://stackoverflow.com/questions/3843363/thread-interrupt-not-ending-blocking-call-on-input-stream-read