From 8ead6a7f6e06d1a68203d699de1faca5bb92e588 Mon Sep 17 00:00:00 2001
From: lou lecrivain <lou.lecrivain@wdz.de>
Date: Thu, 17 Sep 2026 17:21:28 +0200
Subject: [PATCH 1/2] add re-delimiter-codec + regex as dep (needed for chunk
 pre-parsing)

---
 Cargo.lock | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++
 Cargo.toml |  2 ++
 2 files changed, 52 insertions(+)

diff --git a/Cargo.lock b/Cargo.lock
index 08b9478..24c6bb1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2,6 +2,15 @@
 # It is not intended for manual editing.
 version = 4
 
+[[package]]
+name = "aho-corasick"
+version = "1.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "bytes"
 version = "1.12.1"
@@ -38,6 +47,8 @@ version = "0.1.2"
 dependencies = [
  "pest",
  "pest_derive",
+ "re-delimiter-codec",
+ "regex",
  "tokio",
  "tokio-stream",
  "tokio-test",
@@ -110,6 +121,45 @@ dependencies = [
  "proc-macro2",
 ]
 
+[[package]]
+name = "re-delimiter-codec"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bd265cf799a0cdc9eb1417b1403990fd29aafe6992877c70f2b75d1439125e77"
+dependencies = [
+ "regex",
+ "tokio-util",
+]
+
+[[package]]
+name = "regex"
+version = "1.13.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f020237b6c8eed93db2e2cb53c00c60a8e1bc73da7d073199a1180401450218d"
+dependencies = [
+ "aho-corasick",
+ "memchr",
+ "regex-automata",
+ "regex-syntax",
+]
+
+[[package]]
+name = "regex-automata"
+version = "0.4.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2"
+dependencies = [
+ "aho-corasick",
+ "memchr",
+ "regex-syntax",
+]
+
+[[package]]
+name = "regex-syntax"
+version = "0.8.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4"
+
 [[package]]
 name = "syn"
 version = "2.0.119"
diff --git a/Cargo.toml b/Cargo.toml
index bc84ec1..c3c0b25 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -15,6 +15,8 @@ pest_derive = "2.9.0"
 tokio = { version = "1.53.1", optional = true }
 tokio-util = { version = "0.7.19", optional = true, features = ["codec"] }
 tokio-stream = { version = "0.1.19", optional = true }
+re-delimiter-codec = "0.1.0"
+regex = "1.13.1"
 
 
 [features]

From adcbccfb9422cf24ebe552c3d823ed8e2e58d1f1 Mon Sep 17 00:00:00 2001
From: lou lecrivain <lou.lecrivain@wdz.de>
Date: Thu, 17 Sep 2026 18:26:57 +0200
Subject: [PATCH 2/2] WIP implementation for chunk -> guesstimate -> parse

---
 src/streaming.rs | 60 +++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 59 insertions(+), 1 deletion(-)

diff --git a/src/streaming.rs b/src/streaming.rs
index 4e80562..7fff365 100644
--- a/src/streaming.rs
+++ b/src/streaming.rs
@@ -1,6 +1,11 @@
 use crate::{NRTMMessage, NRTMParser, NRTMV2Parser, NRTMV3Parser, ParseError};
+use re_delimiter_codec::{REDelimiterCodec, REDelimiterCodecError};
+use regex::bytes::Regex;
+use std::borrow::Cow;
+use tokio::io::AsyncRead;
+use tokio_stream::{Stream, StreamExt};
 use tokio_util::bytes::BytesMut;
-use tokio_util::codec::Decoder;
+use tokio_util::codec::{Decoder, FramedRead};
 
 const MIN_BUFFER_LEN: usize = 8192;
 const MIN_DECODE_LEN: usize = "ADD 1".len();
@@ -10,6 +15,21 @@ pub struct NRTMDec {
     parser: fn(&str) -> Result<NRTMMessage, ParseError>,
 }
 
+pub enum NRTMReaderError {
+    REDelimiterCodec(REDelimiterCodecError),
+    Parser(ParseError),
+}
+
+fn new_rpsl_preparser() -> REDelimiterCodec {
+    const MAX_CHUNK_LEN: usize = 131072; // 128k
+
+    // ok to call unwrap here, we know this will not fail
+    REDelimiterCodec::new_with_max_length(
+        Regex::new("(?R)\n[^%][^AD][^DE][^DL].*\n\n").unwrap(),
+        MAX_CHUNK_LEN,
+    )
+}
+
 impl NRTMDec {
     pub fn new_v2() -> Self {
         NRTMDec {
@@ -22,6 +42,44 @@ impl NRTMDec {
             parser: NRTMV3Parser::try_parse,
         }
     }
+    pub fn get_stream<T: AsyncRead>(
+        &mut self,
+        reader: T,
+    ) -> impl Stream<Item = Result<NRTMMessage, NRTMReaderError>> {
+        let framed_reader = FramedRead::new(reader, new_rpsl_preparser());
+
+        framed_reader
+            .then(
+                // charset guesstimation
+                async |chunk| {
+                    match chunk {
+                        Ok(chunk) => {
+                            // todo implement charset guesstimation
+                            // 1. guesstimate charset of chunk
+
+                            // 2a. if internally consistent, then decode with charset
+                            // and send cow str. most cases no conversion will
+                            // happen and so cow will stay the same (=speed)
+
+                            // 2b. if internally inconsistent, then proceed to
+                            // lossy utf8 conversion and send back cow str. copy
+                            // will happen but as it is already internally inconsistent
+                            // copy would have been needed anyway
+
+                            Ok(Cow::from(""))
+                        },
+                        Err(e) => Err(e),
+                    }
+                },
+            )
+            .then(async |cow_str| match cow_str {
+                Ok(cow_str) => match (self.parser)(cow_str.as_ref()) {
+                    Ok(message) => Ok(message),
+                    Err(e) => Err(NRTMReaderError::Parser(e)),
+                },
+                Err(e) => Err(NRTMReaderError::REDelimiterCodec(e)),
+            })
+    }
 }
 
 impl Decoder for NRTMDec {