Привожу ниже модуль собственного изготовления. Называется CommAction. Знаю, что залез в чужую подсистему, но слишком уж он мне нравится именно там. Свою изобретать ради двух модулей не стал.

## Код:
```
(*
Общая идея реализации заключается в том, что есть некий Action, который в фоновом режиме опрашивает поток (CommStreams.Stream) на предмет наличия новых данных, получает эти данные и складывает в буфер приема. У этого Action есть также буфер передачи, из которого данные записываются в поток. Интерпретация получаемых данных - дело специфическое, зависящее от конкретных задач, а вот периодическая работа с буферами никаких новшеств не требует. Поэтому Action - не расширяемый тип. Более того, через него невозможно получить прямого доступа ни к потоку, ни к буферам.
Для интерпретации принимаемых данных вводится тип Engine ("движок"). Движок прикрепляется к Action методом Action.SetEngine. Engine - тип расширяемый. Его метод Do опубликован только для реализации и изначально пуст. Этот метод периодически вызывается из Action. Для каждой конкретной задачи следует реализовать свой движок и присоединять его к Action. В общем случае движок реализует некую базовую часть протокола обмена и передает данные далее в протокол верхнего уровня. Движок имеет доступ только к буферам приема и передачи, но не к потоку непосредственно (если иное не предусмотрено в расширенном типе-наследнике). Чтобы подать сигнал завершения работы, движок устанавливает собственное свойство finished = TRUE.
Буферы реализованы в отдельном модуле как тип CommBuffers.Carrier. В текущей реализации они допускают последовательное чтение данных с начала, удаление данных с начала и добавление данных в конец (FIFO).
Таким образом, получение, хранение и обработка данных максимально отделены друг от друга. Объект, обрабатывающий данные, не имеет доступа к коммуникационному потоку, что позволяет четко ограничить доступ к низкоуровневым функциям. Изменяемая логика обработки вынесена в отдельный тип, минимизируя наследуемую часть. Тем самым соблюден принцип предпочтения композиции объектов перед наследованием.
*)

MODULE CommAction;
(* Copyright (c) Alexander Iljin, 2006. *)

IMPORT CommStreams, Services, CommBuffer;

TYPE
   (** Action - это объект, периодически читающий данные из потока в буфер приема и пытающийся записать данные из буфера отправки в поток. Сам по себе Action не предоставляет доступа к этим буферам, для чтения и записи необходимо расширять тип Engine. **)
   Action* = POINTER TO ABSTRACT RECORD (Services.Action)
      active-: BOOLEAN;   (* = TRUE, если действие периодически вызывается *)
      lastRecvTick-, lastSendTick-: LONGINT;   (* когда был последний прием и передача *)
      engine-: Engine   (* движок-интерпретатор получаемых данных *)
   END;

   (** Engine - это объект, который можно присоединить к Action. Action будет периодически вызывать метод Engine.Do для обработки принятых данных или таймаутов. Движок имеет доступ к буферам приема и передачи, но не к потоку непосредственно. Когда работа закончена, движок может установить finished = TRUE и Action закроет поток. **)
   Engine* = POINTER TO EXTENSIBLE RECORD
      action-: Action;   (* с этим Action связан данный движок *)
      recvBuf-, sendBuf-: CommBuffer.Carrier; (* буферы приема и передачи (= NIL тогда и только тогда, когда action = NIL) *)
      finished*: BOOLEAN;   (* Движок может установить finished = TRUE, сигнализируя, что он закончил работу и поток можно закрыть. *)
      streamClosed-: BOOLEAN   (* Action устанавливает streamClosed = TRUE перед вызовом Engine.Do, если поток уже закрыт (например, из-за потери связи). Action все равно будет продолжать работу, пока движок не установит finished = TRUE. *)
   END;

   StdAction = POINTER TO RECORD (Action)
      stream: CommStreams.Stream;   (* коммуникации через этот объект *)
      sendBuf: CommBuffer.Carrier;   (* буфер отправляемых данных *)
      recvBuf: CommBuffer.Carrier;   (* буфер принятых данных *)
      sendRider: CommBuffer.Rider   (* райдер по sendBuf *)
   END;


(* Engine *)

PROCEDURE (e: Engine) Do- (), NEW, EMPTY;
(** Данное событие периодически вызывается, если движок присоединен к Action. Здесь должна выполняться вся работа по интерпретации получаемых данных и отправке данных в ответ. Do вызывается даже если в буфере приема нет новых данных, поэтому можно проверять таймауты относительно Action.lastRecvTick и Action.lastSendTick. **)

PROCEDURE (e: Engine) Close- (), NEW, EMPTY;
(** Данное событие вызывается при нормальном прекращении работы Action. Не будет вызван, если Action просто удалят из списка активных объектов с помощью Services.RemoveAction. **)


PROCEDURE (e: Engine) SetLoggers* (send, recv: CommBuffer.Logger), NEW;
(** Устанавливаем "журналистов" для буферов приема и передачи. **)
BEGIN
   e.recvBuf.SetLogger(recv);
   e.sendBuf.SetLogger(send);
   ASSERT(e.sendBuf.logger = send, 60);
   ASSERT(e.recvBuf.logger = recv, 61)
END SetLoggers;



(* Action *)

PROCEDURE (a: Action) SetEngine* (e: Engine), NEW, ABSTRACT;
(** Задаем движок, который будет обрабатывать данные. *)

PROCEDURE (a: Action) SetStream* (stream: CommStreams.Stream), NEW, ABSTRACT;
(** Задаем поток, через который будем перекачивать данные. *)

PROCEDURE (a: Action) Close*, NEW, ABSTRACT;
(** Закрываем активное соединение и останавливаем движок. *)



(* StdAction *)

(* forward declaration *)
PROCEDURE ^ (a: StdAction) Close*;


PROCEDURE (a: StdAction) SetEngine* (e : Engine);
BEGIN
   (* отцепляем прежний движок *)
   IF a.engine # NIL THEN
      a.engine.action := NIL;
      a.engine.recvBuf := NIL;
      a.engine.sendBuf := NIL
   END;
   (* ставим новый *)
   a.engine := e;
   IF e # NIL THEN
      e.action := a;
      e.recvBuf := a.recvBuf;
      e.sendBuf := a.sendBuf
   END
END SetEngine;


PROCEDURE (a: StdAction) Init, NEW;
BEGIN
   NEW(a.recvBuf);
   NEW(a.sendBuf)
END Init;


PROCEDURE (a: StdAction) SetStream* (stream: CommStreams.Stream);
(** Задаем поток, с которым будем работать. *)
BEGIN
   IF stream = a.stream THEN RETURN END;

   IF a.stream # NIL THEN
      a.stream.Close
(*      ;Log.String("Action: stream closed by setting another stream");
      Log.Ln;*)
   END;
   a.stream := stream;
   IF a.stream # NIL THEN
      Services.DoLater(a, Services.now);
      a.active := TRUE
   END
END SetStream;


PROCEDURE (a: StdAction) Do-;
(* Будем отправлять данные в движок до тех пор, пока буфер приема не будет пуст, даже после того, как сервер закрыл соединение. Это дает возможность движку обрабатывать приходящие данные небольшими порциями, в духе кооперативной многозадачности, и удалять их из буфера по мере обработки, не заводя собственного буфера. *)
   CONST BufSize = 256;
   VAR
      i, num : INTEGER;
      buf : ARRAY BufSize OF BYTE;
      s : POINTER TO ARRAY OF SHORTCHAR;
      ch : SHORTCHAR;
BEGIN
   IF (a.lastSendTick = 0) & (a.lastRecvTick = 0) THEN
      a.lastSendTick := Services.Ticks();
      a.lastRecvTick := Services.Ticks()
   END;

   (* читаем все данные из потока в буфер приема *)
   IF a.stream # NIL THEN
      REPEAT
         a.stream.ReadBytes(buf, 0, BufSize, num);
         a.recvBuf.Append(buf, 0, num);
         IF num # 0 THEN
            a.lastRecvTick := Services.Ticks()
         END
(*         FOR i := 0 TO num-1 DO
            Log.Char(CHR(buf[i]));
         END;*)
      UNTIL num = 0;
   END;

   (* Дадим движку обработать принятые данные. Он может что-то добавить в буфер передачи. *)
   IF (a.engine # NIL) & ~a.engine.finished THEN
      a.engine.streamClosed := (a.stream = NIL) OR ~a.stream.IsConnected();
      a.engine.Do;
   END;

   (* Отправляем данные из буфера передачи в поток. *)
   IF (a.stream # NIL) & a.stream.IsConnected() THEN
      a.sendRider := a.sendBuf.NewRider(a.sendRider);
      IF ~a.sendRider.Eof() THEN
         a.sendRider.ReadBytes(buf, 0, BufSize, num);
         a.stream.WriteBytes(buf, 0, num, num);
         a.sendBuf.Delete(num);
         IF num # 0 THEN
            a.lastSendTick := Services.Ticks()
         END
      END
   END;

   (* Работа продолжается в следующих случаях:
   1. связь установлена, и буфер отправки не пуст, и движок не закончил работу;
   2. если движок не закончил работу и буфер приема не пуст.
   Из пункта 1 следует, что чтобы отправить что-либо "на прощание", следует отсоединить движок от Action и не устанавливать finished. Установка finished = TRUE при наличии движка однозначно трактуется как требование закрыть соединение, даже если буфер отправки не пуст.
    *)
   IF (
      (a.stream # NIL) & a.stream.IsConnected() &
      ( ((a.engine # NIL) & ~a.engine.finished) OR ~a.sendBuf.IsEmpty() )
   )
   OR ((a.engine # NIL) & ~a.recvBuf.IsEmpty() & ~a.engine.finished)
   THEN (* переходим в режим ожидания *)
      Services.DoLater(a, Services.now)
   ELSE (* закрываем соединение *)
      a.Close
   END
END Do;


PROCEDURE (a: StdAction) Close*;
BEGIN
   IF a.stream # NIL THEN
      a.stream.Close;
      a.stream := NIL;
(*      Log.String("stream closed normally");
      Log.Ln;*)
   END;
   Services.RemoveAction(a);
   a.active := FALSE;
   IF a.engine # NIL THEN
      a.engine.Close
   END
END Close;


PROCEDURE (a: StdAction) FINALIZE-;
BEGIN
   a.Close
END FINALIZE;



PROCEDURE NewAction* (stream: CommStreams.Stream; engine: Engine): Action;
   VAR a: StdAction;
BEGIN
   NEW(a);
   a.Init;
   a.SetStream(stream);
   a.SetEngine(engine);
   RETURN a
END NewAction;

END CommAction.
```
