структуры для представления обработки данных в виде конвейера

Большую часть обработки данных можно представить как конвейер компонентов, выход одного подается на вход другого. Типичный конвейер обработки:

reader | handler | writer

В качестве фольги для начала этого обсуждения давайте рассмотрим объектно-ориентированную реализацию этого конвейера, где каждый сегмент является объектом.Объект обработчика содержит ссылки на объекты reader и writer , а также имеет метод run , который выглядит следующим образом:

define handler.run:
  while (reader.has_next) {
    data = reader.next
    output = ...some function of data...
    writer.put(output)
  }

Схематично зависимости:

reader <- handler -> writer

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

reader | tweaker | handler | writer

Опять же, в этой OO-реализации твикер будет оберткой вокруг считывателя , а методы твикера могут выглядеть примерно так (в некотором псевдоимперативном коде):

define tweaker.has_next:
  return reader.has_next

define tweaker.next:
  value = reader.next
  result = ...some function of value...
  return result

Я считаю, что это не очень компонуемая абстракция. Некоторые проблемы:

1. твикер может использоваться только в левой части обработчика , т.е. я не могу использовать вышеупомянутую реализацию твикера для формирования этого конвейера. :

   читатель | обработчик | твикер | Writer

2. Я хотел бы использовать ассоциативное свойство конвейеров, чтобы этот конвейер:

   reader | обработчик | писатель

может быть выражен как:

reader | p

где p - конвейер обработчик | писатель . В этой объектно-ориентированной реализации мне пришлось бы частично создать экземпляр объекта обработчика

1. Что-то вроде повторения (1), объекты должны знать, «выталкивают» они или «вытягивают» данные.

Я ищу структуру (не обязательно объектно-ориентированную) для создания конвейеров обработки данных, которые решают эти проблемы.

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

В качестве цели было бы неплохо создать такой конвейер:

                     handler1
                   /          \
reader | partition              writer
                   \          /
                     handler2

С некоторой точки зрения, конвейеры оболочки Unix решают многие из этих проблем с помощью следующих решений по реализации:

1. Компоненты конвейера работают асинхронно в отдельных процессах

2. Объекты трубы служат посредником при передаче данных между «толкателями» и «съемниками»; т.е. они блокируют писателей, которые пишут данные слишком быстро, и читателей, которые пытаются читать слишком быстро.

3. Вы используете специальные соединители < и > для подключения пассивных компонентов (то есть файлов) к конвейеру

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

Спасибо!