diff --git a/config/config.go b/config/config.go
index 9c35cf26626e0..b116df2f44012 100644
--- a/config/config.go
+++ b/config/config.go
@@ -1160,6 +1160,23 @@ func (c *Config) addParser(parentcategory, parentname string, table *ast.Table)
 		}
 	}
 
+	// Handle embedded parsers for parsers that support them
+	if t, ok := parser.(telegraf.ParserPlugin); ok {
+		val, found := table.Fields["embedded_parser"]
+		if !found {
+			return nil, fmt.Errorf("parser %q requires an 'embedded_parser' table", conf.DataFormat)
+		}
+		subTable, ok := val.(*ast.Table)
+		if !ok {
+			return nil, errors.New("invalid 'embedded_parser' table")
+		}
+		embedded, err := c.addParser(parentcategory, conf.DataFormat, subTable)
+		if err != nil {
+			return nil, fmt.Errorf("adding embedded parser failed: %w", err)
+		}
+		t.SetParser(embedded)
+	}
+
 	if err := c.toml.UnmarshalTable(table, parser); err != nil {
 		return nil, err
 	}
@@ -1824,6 +1841,9 @@ func (c *Config) missingTomlField(_ reflect.Type, key string) error {
 	// Parser and serializer options to ignore
 	case "data_type", "influx_parser_type":
 
+	// Options handled separately when building embedded parsers
+	case "embedded_parser":
+
 	default:
 		c.unusedFieldsMutex.Lock()
 		c.UnusedFields[key] = true
diff --git a/plugins/parsers/all/concat.go b/plugins/parsers/all/concat.go
new file mode 100644
index 0000000000000..8d8e2f4774353
--- /dev/null
+++ b/plugins/parsers/all/concat.go
@@ -0,0 +1,5 @@
+//go:build !custom || parsers || parsers.concat
+
+package all
+
+import _ "github.com/influxdata/telegraf/plugins/parsers/concat" // register plugin
diff --git a/plugins/parsers/concat/README.md b/plugins/parsers/concat/README.md
new file mode 100644
index 0000000000000..cb7b410ed7b04
--- /dev/null
+++ b/plugins/parsers/concat/README.md
@@ -0,0 +1,120 @@
+# Concat Parser Plugin
+
+The `concat` parser splits a binary byte stream into messages by locating and
+validating a message header, then passes each message to an embedded parser
+(such as `json`, `csv` or `binary`) to produce metrics.
+
+Use it for framed binary protocols where a header marks the start of each
+message. Validating the header lets the parser find message boundaries reliably
+even when reception starts mid-message or a message is spread across several
+packets (e.g. UDP datagrams).
+
+For each chunk of input the parser buffers the bytes and tries to read any
+complete messages. If a header does not validate, the parser advances one byte
+and retries. If the header validates but the message is incomplete, the parser
+returns no metrics and keeps the buffer until the next chunk arrives.
+
+> [!NOTE]
+> The parser is stateful, keeping a buffer and scan cursor across calls. Run one
+> instance per stream and keep `max_parallel_parsers = 1` (the default);
+> parallel parsing corrupts its state.
+> The parser is susceptible to dropping / creating nonexistent metrics depending
+> on the chosen header format. It is likely that this bogus data will be caught
+> by the embedded parser, but precautions should still be made.
+
+## Configuration
+
+```toml
+[[inputs.socket_listener]]
+  service_address = "udp://:9000"
+
+  ## Data format to consume.
+  ## Each data format has its own unique set of configuration options, read
+  ## more about them here:
+  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
+  data_format = "concat"
+
+  ## The parser is stateful and must not run in parallel.
+  max_parallel_parsers = 1
+
+  ## Header schema used to locate and validate message boundaries.
+  [inputs.socket_listener.split]
+    ## Bytes that are constant in every header, as offset = "hex" pairs
+    ## (a "0x"/"x" prefix is optional). Used to spot the start of a message.
+    const_bytes = {0 = "0x50"}
+
+    ## Number of header bytes to strip before passing a message to the
+    ## embedded parser. If zero, the header is kept.
+    header_length = 5
+
+    ## Field holding the total message length in bytes (header included).
+    ## If omitted, a message runs until the start of the next valid header.
+    [inputs.socket_listener.split.message_length_field]
+      offset = 2
+      bytes = 2
+      endianness = "be"  # "be" or "le"
+      max_length = 0     # reject a header whose length exceeds this; 0 = no limit
+
+    ## Checksum verifying a candidate header. A mismatch triggers
+    ## resynchronization. Offsets and range are relative to the header start.
+    [inputs.socket_listener.split.checksum]
+      strategy = "xor"  # "none", "crc32", "md5", "sha256" or "xor"
+      range = [0, 3]    # half-open [start, end) covered by the checksum
+      offset = 4        # where the checksum is stored
+      bytes = 1         # crc32=4, md5=16, sha256=32; xor is the fold width
+
+  ## Embedded parser applied to each extracted message, selected by its own
+  ## data_format and configured like any other parser.
+  [inputs.socket_listener.embedded_parser]
+    data_format = "json"
+```
+
+### message_length_field
+
+An unsigned integer (width 1, 2, 4 or 8 bytes) giving the **total** message
+length, header included. The parser consumes that many bytes and strips
+`header_length` of them before parsing the remainder.
+
+Set `max_length` to bound the accepted length. A header reporting a length
+greater than `max_length` is treated as invalid and the parser resynchronizes.
+
+Omit this field to run in delimiter mode, where a message ends at the start of
+the next valid header. Delimiter mode needs `const_bytes` and/or a `checksum` so
+headers can be recognized. In delimiter mode a message is only emitted once the
+next one starts.
+
+### checksum
+
+The checksum is computed over the half-open `range` `[start, end)` and compared
+against the `bytes`-wide value stored at `offset`. A mismatch is treated like a
+failed header match, so the parser resynchronizes.
+
+## Example
+
+The sample configuration above describes a 5-byte header followed by a
+JSON payload:
+
+```text
+ offset    0      1      2      3      4      5 ...
+        +------+------+------+------+------+-------------------+
+        | 0x50 | seq  |   length    | xor  |  payload (JSON)   |
+        +------+------+------+------+------+-------------------+
+         const             uint16    check
+```
+
+- `const_bytes = {0 = "0x50"}` — byte 0 is always `0x50`, marking a header.
+- `message_length_field` at offset 2, 2 bytes big-endian — the `length` field
+  holds the total frame size, so a `length` of `20` means 15 bytes of payload.
+- `checksum` `xor` over `range = [0, 3]` stored at `offset = 4` — the parser
+  XORs bytes 0-2 and compares the result against byte 4 to confirm the header.
+- `header_length = 5` — the five header bytes are stripped before the `length`-5
+  payload bytes are handed to the embedded JSON parser.
+
+Byte 1 (`seq`) is not referenced by any option, so it is neither validated nor
+stripped separately; it is simply part of the discarded header.
+
+## Behavior
+
+- Bytes skipped during resynchronization are dropped.
+- In length mode, if a valid header appears before the length-implied end, the
+  message is treated as truncated and discarded up to that header.
diff --git a/plugins/parsers/concat/concat.go b/plugins/parsers/concat/concat.go
new file mode 100644
index 0000000000000..9d14cf8ef3d53
--- /dev/null
+++ b/plugins/parsers/concat/concat.go
@@ -0,0 +1,67 @@
+package concat
+
+import (
+	"bytes"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/parsers"
+)
+
+type Parser struct {
+	Splitter Splitter `toml:"split"`
+
+	parser telegraf.Parser
+	buffer bytes.Buffer
+}
+
+// SetParser sets the embedded parser applied to each extracted message. It
+// implements telegraf.ParserPlugin so the config can build the parser from the
+// embedded_parser sub-table.
+func (p *Parser) SetParser(parser telegraf.Parser) {
+	p.parser = parser
+}
+
+func (p *Parser) Parse(buf []byte) ([]telegraf.Metric, error) {
+	p.buffer.Write(buf)
+	var metrics []telegraf.Metric
+	for {
+		advance, msg, err := p.Splitter.Split(p.buffer.Bytes(), false)
+		if err != nil {
+			return metrics, err
+		}
+		if advance == 0 {
+			// Not enough data for a complete message yet.
+			break
+		}
+		if msg != nil {
+			m := make([]byte, len(msg))
+			copy(m, msg)
+			metricsPart, err := p.parser.Parse(m)
+			if err != nil {
+				p.buffer.Next(advance)
+				return metrics, err
+			}
+			metrics = append(metrics, metricsPart...)
+		}
+		p.buffer.Next(advance)
+	}
+	return metrics, nil
+}
+
+func (p *Parser) ParseLine(line string) (telegraf.Metric, error) {
+	return p.parser.ParseLine(line)
+}
+
+func (p *Parser) SetDefaultTags(tags map[string]string) {
+	p.parser.SetDefaultTags(tags)
+}
+
+func (p *Parser) Init() error {
+	return p.Splitter.Init()
+}
+
+func init() {
+	parsers.Add("concat", func(string) telegraf.Parser {
+		return &Parser{}
+	})
+}
diff --git a/plugins/parsers/concat/concat_test.go b/plugins/parsers/concat/concat_test.go
new file mode 100644
index 0000000000000..b1880aa3b5d42
--- /dev/null
+++ b/plugins/parsers/concat/concat_test.go
@@ -0,0 +1,252 @@
+package concat
+
+import (
+	"bytes"
+	mathrand "math/rand"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/metric"
+)
+
+// newRand returns a deterministically seeded RNG so test runs are reproducible.
+// The seed is logged so a failing run can be replayed.
+func newRand(t *testing.T) *mathrand.Rand {
+	t.Helper()
+	const seed = 42
+	t.Logf("random seed: %d", seed)
+	return mathrand.New(mathrand.NewSource(seed))
+}
+
+func randomBytes(t *testing.T, rng *mathrand.Rand, n int) []byte {
+	t.Helper()
+	b := make([]byte, n)
+	_, err := rng.Read(b)
+	require.NoError(t, err)
+	return b
+}
+
+func randomMessages(t *testing.T, rng *mathrand.Rand, n, minLen, maxLen int, constBytes []byte) [][]byte {
+	t.Helper()
+	messages := make([][]byte, n)
+	for i := 0; i < n; i++ {
+		length := minLen + rng.Intn(maxLen-minLen+1) + 6
+		lenbytes := make([]byte, 2)
+		lenbytes[0] = uint8(length >> 8)
+		lenbytes[1] = uint8(length & 0xff)
+		messages[i] = append(constBytes, uint8(i), lenbytes[0], lenbytes[1], 0)
+		checksum := uint8(0)
+		for j := 0; j < 5; j++ {
+			checksum ^= messages[i][j]
+		}
+		messages[i] = append(messages[i], checksum)
+		messages[i] = append(messages[i], randomBytes(t, rng, length-6)...)
+	}
+	return messages
+}
+
+// rawParser is a minimal embedded parser that wraps each message payload in a
+// metric so tests can verify the bytes the splitter handed over.
+type rawParser struct{}
+
+func (*rawParser) Parse(buf []byte) ([]telegraf.Metric, error) {
+	m := metric.New("raw", nil, map[string]interface{}{"payload": string(buf)}, time.Unix(0, 0))
+	return []telegraf.Metric{m}, nil
+}
+
+func (*rawParser) ParseLine(line string) (telegraf.Metric, error) {
+	return metric.New("raw", nil, map[string]interface{}{"payload": line}, time.Unix(0, 0)), nil
+}
+
+func (*rawParser) SetDefaultTags(map[string]string) {}
+
+func payload(t *testing.T, m telegraf.Metric) string {
+	t.Helper()
+	v, ok := m.GetField("payload")
+	require.True(t, ok, "metric is missing the payload field")
+	s, ok := v.(string)
+	require.True(t, ok, "payload field is not a string")
+	return s
+}
+
+func TestConcatParserDelim(t *testing.T) {
+	p := &Parser{
+		Splitter: Splitter{
+			HeaderLength: 6,
+			MessageLengthField: LengthFieldConfig{
+				Offset:     2,
+				Bytes:      0,
+				Endianness: "be",
+			},
+			ConstBytes: map[string]string{
+				"0": "0x73",
+				"1": "0x74",
+				"2": "0x61",
+				"3": "0x72",
+				"4": "0x74",
+			},
+		},
+	}
+	p.SetParser(&rawParser{})
+	require.NoError(t, p.Init())
+
+	rng := newRand(t)
+
+	t.Run("single message", func(t *testing.T) {
+		constBytes := []byte{'s', 't', 'a', 'r', 't'}
+		msgs := randomMessages(t, rng, 2, 10, 20, constBytes)
+		// In delimiter mode a message is only emitted once the next header is
+		// seen, so feed a following message to flush the first.
+		metrics, err := p.Parse(msgs[0])
+		require.NoError(t, err)
+		require.Empty(t, metrics, "expected no metrics before next header")
+
+		metrics, err = p.Parse(msgs[1])
+		require.NoError(t, err)
+		require.Len(t, metrics, 1)
+		require.Equal(t, string(msgs[0][6:]), payload(t, metrics[0]))
+	})
+
+	t.Run("multiple messages", func(t *testing.T) {
+		const (
+			n              = 10000
+			minLen         = 100
+			maxLen         = 2000
+			allowedDropped = 0.1
+			allowedGhost   = 0.1
+		)
+		constBytes := []byte{'s', 't', 'a', 'r', 't'}
+		messages := randomMessages(t, rng, n, minLen, maxLen, constBytes)
+
+		stream := bytes.Join(messages, nil)
+		var metrics []telegraf.Metric
+		for len(stream) > 0 {
+			pkt := rng.Intn(2048) + 1
+			if pkt > len(stream) {
+				pkt = len(stream)
+			}
+			parsed, err := p.Parse(stream[:pkt])
+			require.NoError(t, err)
+			metrics = append(metrics, parsed...)
+			stream = stream[pkt:]
+		}
+		// Reconcile emitted metrics against the source messages. Payloads are
+		// random variable-length blobs.
+		dropped := 0
+		ghostMetrics := 0
+		mi := 0
+		for _, m := range metrics {
+			buf := payload(t, m)
+			matched := false
+			for j := mi; j < len(messages); j++ {
+				if buf == string(messages[j][6:]) {
+					dropped += j - mi
+					mi = j + 1
+					matched = true
+					break
+				}
+			}
+			if !matched {
+				ghostMetrics++
+			}
+		}
+		// Any messages left unmatched after the last metric were also dropped.
+		// In delimiter mode the final message stays buffered until a following
+		// header arrives, so it is expected to show up here.
+		dropped += len(messages) - mi
+
+		t.Logf("out of: %d, dropped metrics: %d, ghost metrics: %d", n, dropped, ghostMetrics)
+		require.LessOrEqualf(t, float64(dropped)/float64(n), allowedDropped,
+			"dropped %d of %d metrics", dropped, n)
+		require.LessOrEqualf(t, float64(ghostMetrics)/float64(n), allowedGhost,
+			"ghost metrics: %d", ghostMetrics)
+	})
+}
+func TestConcatParserLength(t *testing.T) {
+	p := &Parser{
+		Splitter: Splitter{
+			HeaderLength: 6,
+			MessageLengthField: LengthFieldConfig{
+				Offset:     2,
+				Bytes:      2,
+				Endianness: "be",
+			},
+			ConstBytes: map[string]string{
+				"0": "0x53",
+			},
+			Checksum: ChecksumConfig{
+				Strategy: "xor",
+				Range:    []int{0, 4},
+				Offset:   5,
+				Bytes:    1,
+			},
+		},
+	}
+	p.SetParser(&rawParser{})
+	require.NoError(t, p.Init())
+
+	rng := newRand(t)
+
+	t.Run("single message", func(t *testing.T) {
+		constBytes := []byte{'S'}
+		m := randomMessages(t, rng, 1, 10, 20, constBytes)[0]
+		metrics, err := p.Parse(m)
+		require.NoError(t, err)
+		require.Len(t, metrics, 1)
+		require.Equal(t, string(m[6:]), payload(t, metrics[0]))
+	})
+
+	t.Run("multiple messages", func(t *testing.T) {
+		const (
+			n              = 10000
+			minLen         = 100
+			maxLen         = 200
+			allowedDropped = 0.1
+			allowedGhost   = 0.1
+		)
+		messages := randomMessages(t, rng, n, minLen, maxLen, []byte{'S'})
+		stream := bytes.Join(messages, nil)
+		var metrics []telegraf.Metric
+		for len(stream) > 0 {
+			pkt := rng.Intn(2048) + 1
+			if pkt > len(stream) {
+				pkt = len(stream)
+			}
+			parsed, err := p.Parse(stream[:pkt])
+			require.NoError(t, err)
+			metrics = append(metrics, parsed...)
+			stream = stream[pkt:]
+		}
+		// Reconcile emitted metrics against the source messages. Payloads are
+		// random variable-length blobs.
+		dropped := 0
+		ghostMetrics := 0
+		mi := 0
+		for _, m := range metrics {
+			buf := payload(t, m)
+			matched := false
+			for j := mi; j < len(messages); j++ {
+				if buf == string(messages[j][6:]) {
+					dropped += j - mi
+					mi = j + 1
+					matched = true
+					break
+				}
+			}
+			if !matched {
+				ghostMetrics++
+			}
+		}
+		// Any messages left unmatched after the last metric were also dropped.
+		dropped += len(messages) - mi
+
+		t.Logf("out of: %d, dropped metrics: %d, ghost metrics: %d", n, dropped, ghostMetrics)
+		require.LessOrEqualf(t, float64(dropped)/float64(n), allowedDropped,
+			"dropped %d of %d metrics", dropped, n)
+		require.LessOrEqualf(t, float64(ghostMetrics)/float64(n), allowedGhost,
+			"ghost metrics: %d", ghostMetrics)
+	})
+}
diff --git a/plugins/parsers/concat/split.go b/plugins/parsers/concat/split.go
new file mode 100644
index 0000000000000..963d4422c3851
--- /dev/null
+++ b/plugins/parsers/concat/split.go
@@ -0,0 +1,299 @@
+package concat
+
+import (
+	"bufio"
+	"bytes"
+	"crypto/md5"
+	"crypto/sha256"
+	"encoding/binary"
+	"encoding/hex"
+	"fmt"
+	"hash/crc32"
+	"strconv"
+	"strings"
+)
+
+type Splitter struct {
+	ConstBytes         map[string]string `toml:"const_bytes"`
+	HeaderLength       int               `toml:"header_length"`
+	MessageLengthField LengthFieldConfig `toml:"message_length_field"`
+	Checksum           ChecksumConfig    `toml:"checksum"`
+	s                  bufio.SplitFunc
+
+	// Resume cursor within the current message; see the scan loop in Init.
+	scanPos int
+}
+
+type ChecksumConfig struct {
+	Strategy string `toml:"strategy"`
+	Range    []int  `toml:"range"`
+	Offset   int    `toml:"offset"`
+	Bytes    int    `toml:"bytes"`
+}
+
+type LengthFieldConfig struct {
+	Offset     int    `toml:"offset"`
+	Bytes      int    `toml:"bytes"`
+	Endianness string `toml:"endianness"`
+	MaxLength  int    `toml:"max_length"`
+}
+
+func (s *Splitter) Init() error {
+	// Parse the constant bytes into an offset -> value map
+	constBytes := make(map[int]byte, len(s.ConstBytes))
+	for k, v := range s.ConstBytes {
+		offset, err := strconv.Atoi(k)
+		if err != nil {
+			return fmt.Errorf("invalid const byte offset %q: %w", k, err)
+		}
+		if offset < 0 {
+			return fmt.Errorf("const byte offset %d must not be negative", offset)
+		}
+		if s.HeaderLength > 0 && offset >= s.HeaderLength {
+			return fmt.Errorf("const byte offset %d is past the header length %d", offset, s.HeaderLength)
+		}
+		// Decode the hex value, accepting an optional "0x"/"x" prefix.
+		h := strings.ToLower(v)
+		h = strings.TrimPrefix(h, "0x")
+		h = strings.TrimPrefix(h, "x")
+		b, err := hex.DecodeString(h)
+		if err != nil {
+			return fmt.Errorf("invalid const byte value %q at offset %d: %w", v, offset, err)
+		}
+		if len(b) != 1 {
+			return fmt.Errorf("const byte value %q at offset %d is not a single byte", v, offset)
+		}
+		constBytes[offset] = b[0]
+	}
+
+	lengthConv, err := makeLengthConverter(s.MessageLengthField.Bytes, s.MessageLengthField.Endianness)
+	if err != nil {
+		return err
+	}
+	lf := s.MessageLengthField
+
+	checksumFn, err := makeChecksum(s.Checksum.Strategy, s.Checksum.Bytes)
+	if err != nil {
+		return err
+	}
+	cs := s.Checksum
+	if checksumFn != nil {
+		if len(cs.Range) != 2 {
+			return fmt.Errorf("checksum requires a range [start, end]")
+		}
+		if cs.Range[0] < 0 || cs.Range[1] < cs.Range[0] {
+			return fmt.Errorf("invalid checksum range %v", cs.Range)
+		}
+	}
+
+	// Without a length field, boundaries are found by scanning for the next
+	// header, so we need something to recognize a header by.
+	if lf.Bytes <= 0 && len(constBytes) == 0 && checksumFn == nil {
+		return fmt.Errorf("cannot determine message boundaries: configure a length field, constant bytes, or a checksum")
+	}
+
+	// Bytes needed before a header can be validated at a given position.
+	minlen := 0
+	if lf.Bytes > 0 {
+		minlen = max(minlen, lf.Offset+lf.Bytes)
+	}
+	for offset := range constBytes {
+		minlen = max(minlen, offset+1)
+	}
+	if checksumFn != nil {
+		minlen = max(minlen, cs.Offset+cs.Bytes)
+		minlen = max(minlen, cs.Range[1])
+	}
+
+	headerLen := s.HeaderLength
+
+	// headerAt reports whether a valid header starts at position p. The caller
+	// must ensure at least p+minlen bytes are available.
+	headerAt := func(buf []byte, p int) bool {
+		for offset, b := range constBytes {
+			if buf[p+offset] != b {
+				return false
+			}
+		}
+		if checksumFn != nil {
+			sum := checksumFn(buf[p+cs.Range[0] : p+cs.Range[1]])
+			if !bytes.Equal(sum, buf[p+cs.Offset:p+cs.Offset+cs.Bytes]) {
+				return false
+			}
+		}
+		return true
+	}
+
+	// scanPos is 0 at the start of a new message (header not yet validated) and
+	// >0 once validated, when we are only looking for the end. It is reset to 0
+	// on every message boundary.
+	s.scanPos = 0
+
+	if lf.Bytes > 0 {
+		// Length mode: the length field gives the message end.
+		s.s = func(data []byte, _ bool) (advance int, token []byte, err error) {
+			dataLen := len(data)
+			if dataLen == 0 || dataLen < minlen {
+				return 0, nil, nil
+			}
+
+			// Validate the header only at the start of a new message; resync by
+			// one byte on mismatch.
+			if s.scanPos == 0 {
+				if !headerAt(data, 0) {
+					return 1, nil, nil
+				}
+				s.scanPos = 1
+			}
+
+			end := lengthConv(data[lf.Offset : lf.Offset+lf.Bytes])
+			if end <= 0 || (lf.MaxLength > 0 && end > lf.MaxLength) {
+				s.scanPos = 0
+				return 1, nil, nil
+			}
+			// A valid header before the claimed end means the length is bogus
+			// (desync or truncation); discard up to that header.
+			p := s.scanPos
+			for ; p < end && p+minlen <= dataLen; p++ {
+				if headerAt(data, p) {
+					s.scanPos = 0
+					return p, nil, nil
+				}
+			}
+			if end > dataLen {
+				s.scanPos = p
+				return 0, nil, nil
+			}
+			s.scanPos = 0
+			if end > headerLen {
+				return end, data[headerLen:end], nil
+			}
+			// The length field is too small to include any payload; discard it.
+			return end, nil, nil
+		}
+		return nil
+	}
+
+	// Delimiter mode: the message ends at the next valid header.
+	s.s = func(data []byte, _ bool) (advance int, token []byte, err error) {
+		dataLen := len(data)
+		if dataLen == 0 || dataLen < minlen {
+			return 0, nil, nil
+		}
+
+		// Validate the header only at the start of a new message; resync by
+		// one byte on mismatch.
+		if s.scanPos == 0 {
+			if !headerAt(data, 0) {
+				return 1, nil, nil
+			}
+			s.scanPos = 1
+		}
+
+		// A header can only start after the current message's header, so never
+		// look before headerLen. Resume from scanPos so each position is
+		// examined once.
+		p := s.scanPos
+		if p < headerLen {
+			p = headerLen
+		}
+		for ; p+minlen <= dataLen; p++ {
+			if headerAt(data, p) {
+				s.scanPos = 0
+				return p, data[headerLen:p], nil
+			}
+		}
+		s.scanPos = p
+		return 0, nil, nil
+	}
+
+	return nil
+}
+
+// Split implements bufio.SplitFunc, returning one message with the header
+// stripped. An (advance>0, nil) return means bytes were skipped to
+// resynchronize without producing a message.
+func (s *Splitter) Split(data []byte, atEOF bool) (advance int, token []byte, err error) {
+	if s.s == nil {
+		return 0, nil, nil
+	}
+	return s.s(data, atEOF)
+}
+
+// makeLengthConverter returns a function decoding a big- or little-endian
+// unsigned integer of the given width. It returns nil when no length field is
+// configured (nbytes <= 0).
+func makeLengthConverter(nbytes int, endianness string) (func([]byte) int, error) {
+	if nbytes <= 0 {
+		return nil, nil
+	}
+	var order binary.ByteOrder
+	switch strings.ToLower(endianness) {
+	case "", "be":
+		order = binary.BigEndian
+	case "le":
+		order = binary.LittleEndian
+	default:
+		return nil, fmt.Errorf("invalid endianness %q", endianness)
+	}
+	switch nbytes {
+	case 1:
+		return func(b []byte) int { return int(b[0]) }, nil
+	case 2:
+		return func(b []byte) int { return int(order.Uint16(b)) }, nil
+	case 4:
+		return func(b []byte) int { return int(order.Uint32(b)) }, nil
+	case 8:
+		return func(b []byte) int { return int(order.Uint64(b)) }, nil
+	default:
+		return nil, fmt.Errorf("invalid length field width %d", nbytes)
+	}
+}
+
+// makeChecksum returns a function computing the checksum of the given data
+// using the requested strategy, producing a digest of width bytes. It returns
+// nil when no checksum is configured.
+func makeChecksum(strategy string, width int) (func([]byte) []byte, error) {
+	switch strings.ToLower(strategy) {
+	case "", "none":
+		return nil, nil
+	case "crc32":
+		if width != 4 {
+			return nil, fmt.Errorf("crc32 checksum requires 4 bytes, got %d", width)
+		}
+		return func(data []byte) []byte {
+			b := make([]byte, 4)
+			binary.BigEndian.PutUint32(b, crc32.ChecksumIEEE(data))
+			return b
+		}, nil
+	case "md5":
+		if width != md5.Size {
+			return nil, fmt.Errorf("md5 checksum requires %d bytes, got %d", md5.Size, width)
+		}
+		return func(data []byte) []byte {
+			sum := md5.Sum(data)
+			return sum[:]
+		}, nil
+	case "sha256":
+		if width != sha256.Size {
+			return nil, fmt.Errorf("sha256 checksum requires %d bytes, got %d", sha256.Size, width)
+		}
+		return func(data []byte) []byte {
+			sum := sha256.Sum256(data)
+			return sum[:]
+		}, nil
+	case "xor":
+		if width <= 0 {
+			return nil, fmt.Errorf("xor checksum requires a positive byte width")
+		}
+		return func(data []byte) []byte {
+			sum := make([]byte, width)
+			for i, b := range data {
+				sum[i%width] ^= b
+			}
+			return sum
+		}, nil
+	default:
+		return nil, fmt.Errorf("unknown checksum strategy %q", strategy)
+	}
+}