Repository: meltano/sdk
Pull request #2541 (2024-07-17 01:48:26)

feat: Implement Singer msgspec encoding
- [x] Docs: https://meltano-sdk--2541.org.readthedocs.build/en/2541/guides/performance.html
- [x] UX. Relying on mro magic is not great, e.g. why `SQLiteTap(MsgSpecWriter, SQLTap)` and not `SQLiteTap(SQLTap, MsgSpecWriter)`. A better approach might be to make the IO implementation an attribute of the Singer class.
- [x] Fix tests

<!-- readthedocs-preview meltano-sdk start -->
----
📚 Documentation preview 📚: https://meltano-sdk--2541.org.readthedocs.build/en/2541/

<!-- readthedocs-preview meltano-sdk end -->

## Summary by Sourcery

Implement `msgspec` encoding for improved performance.

Enhancements:
- Replace the default JSON encoder and decoder with `msgspec` for serialization and deserialization.

Tests:
- Update tests to accommodate the `msgspec` implementation.

codspeed-hq[bot]: ## [CodSpeed Performance Report](https://codspeed.io/meltano/sdk/branches/edgarrmondragon/refactor/msgspec-impl-naive)

### Merging #2541 will **improve performances by ×13**

<sub>Comparing <code>edgarrmondragon/refactor/msgspec-impl-naive</code> (97de9f7) with <code>main</code> (32c059b)</sub>



### Summary

`⚡ 2` improvements
`✅ 4` untouched benchmarks




### Benchmarks breakdown

|     | Benchmark | `main` | `edgarrmondragon/refactor/msgspec-impl-naive` | Change |
| --- | --------- | ----------------------- | ------------------- | ------ |
| ⚡ | `test_bench_deserialize_json` | 24.2 ms | 5.5 ms | ×4.4 |
| ⚡ | `test_bench_format_message` | 53.5 ms | 4.1 ms | ×13 |

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/meltano/sdk/pull/2541?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) Report
Attention: Patch coverage is `71.87500%` with `9 lines` in your changes missing coverage. Please review.
> Project coverage is 89.29%. Comparing base [(`32c059b`)](https://app.codecov.io/gh/meltano/sdk/commit/32c059bfd3de23705e70fb44be3278c931383832?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) to head [(`97de9f7`)](https://app.codecov.io/gh/meltano/sdk/commit/97de9f79e4862444e196cac6beafc5ea2f3b13a0?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano).

| [Files](https://app.codecov.io/gh/meltano/sdk/pull/2541?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) | Patch % | Lines |
|---|---|---|
| [singer\_sdk/\_singerlib/encoding/\_msgspec.py](https://app.codecov.io/gh/meltano/sdk/pull/2541?src=pr&el=tree&filepath=singer_sdk%2F_singerlib%2Fencoding%2F_msgspec.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano#diff-c2luZ2VyX3Nkay9fc2luZ2VybGliL2VuY29kaW5nL19tc2dzcGVjLnB5) | 71.87% | [9 Missing :warning: ](https://app.codecov.io/gh/meltano/sdk/pull/2541?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main    #2541      +/-   ##
==========================================
- Coverage   89.41%   89.29%   -0.12%     
==========================================
  Files          58       59       +1     
  Lines        4781     4813      +32     
  Branches      933      933              
==========================================
+ Hits         4275     4298      +23     
- Misses        353      362       +9     
  Partials      153      153              
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/meltano/sdk/pull/2541?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano).

BuzzCutNorman: I found that it helped to add a `defualt_output` to the for the SingerWriter to use.  This allows you to make the write message a little generic.

```
default_output = sys.stdout.buffer

def write_message(self, message: Message) -> None:
	"""Write a message to stdout.

	Args:
		message: The message to write.
	"""
	self.default_output.write(self.format_message(message))
	self.default_output.flush()
```
BuzzCutNorman: In the `json.py` file I found to match the msgspec performance suggestions and fit into the framework you put in place I created a function for generating jsonl so I could keep the functionality of seralize_json to return strings.  This way the serialize_json can be used in the connector engine creation.

https://jcristharif.com/msgspec/perf-tips.html#line-delimited-json

json.py:
```
def serialize_json(obj: object, **kwargs: t.Any) -> str:
    """Serialize a dictionary into a line of json.

    Args:
        obj: A Python object usually a dict.
        **kwargs: Optional key word arguments.

    Returns:
        A string of serialized json.
    """
    return encoder.encode(obj).decode()

msg_buffer = bytearray(64)

def serialize_jsonl(obj: object, **kwargs: t.Any) -> bytes:
        """Serialize a dictionary into a line of jsonl.

        Args:
            obj: A Python object usually a dict.
            **kwargs: Optional key word arguments.

        Returns:
            A bytes of serialized json.
        """
        encoder.encode_into(obj, msg_buffer)
        msg_buffer.extend(b"\n")
        return msg_buffer

```

SingerWriter:
```
    def serialize_message(self, message: Message) -> str | bytes:
        """Serialize a dictionary into a line of json.

        Args:
            message: A Singer message object.

        Returns:
            A string of serialized json.
        """
        return serialize_jsonl(message.to_dict())
```

edgarrmondragon: > I found that it helped to add a `defualt_output` to the for the SingerWriter to use. This allows you to make the write message a little generic.
> 
> ```
> default_output = sys.stdout.buffer
> 
> def write_message(self, message: Message) -> None:
> 	"""Write a message to stdout.
> 
> 	Args:
> 		message: The message to write.
> 	"""
> 	self.default_output.write(self.format_message(message))
> 	self.default_output.flush()
> ```

Do you mean in

https://github.com/meltano/sdk/blob/08b58bf4601eefc88418e0f696bad253bb7b91c9/singer_sdk/_singerlib/encoding/_msgspec.py#L73-L95

?
BuzzCutNorman: Yes, that is exactly what I meant.  Could have defiantly been stated clearer on my part 😅.  

```
 class MsgSpecWriter(GenericSingerWriter[bytes, Message]): 
     """Interface for all plugins writing Singer messages to stdout.""" 
     
     default_output = sys.stdout.buffer
     
     def serialize_message(self, message: Message) -> bytes:  # noqa: PLR6301 
         """Serialize a dictionary into a line of json. 
```
edgarrmondragon: Naive of me to think I could get this across in 1/2 a day of work 😅. I'll come back to this later, there's plenty of time until the planned release date.
BuzzCutNorman: Like the pun 😊.  Great dad joke material.  Kind an inside joke now since you dropped (naive) from the title of the PR.  
edgarrmondragon: Might have to punt this if https://github.com/jcrist/msgspec/pull/711 doesn't get merged before we're ready to officially declare Python 3.13 support
BuzzCutNorman: > I found that it helped to add a defualt_output to the for the SingerWriter to use. This allows you to make the write message a little generic.
> ```
>  class MsgSpecWriter(GenericSingerWriter[bytes, Message]): 
>      """Interface for all plugins writing Singer messages to stdout.""" 
>      
>      default_output = sys.stdout.buffer
>      
>      def serialize_message(self, message: Message) -> bytes:  # noqa: PLR6301 
>          """Serialize a dictionary into a line of json. 
>   
>          Args: 
>              message: A Singer message object. 
>   
>          Returns: 
>              A string of serialized json. 
>          """ 
>          return serialize_jsonl(message.to_dict()) 
>   
>      def write_message(self, message: Message) -> None: 
>          """Write a message to stdout. 
>   
>          Args: 
>              message: The message to write. 
>          """ 
>  	self.default_output.write(self.format_message(message))
> 	self.default_output.flush()
> ```

Well, found that making this generic by using `default_output` was a bad move.  It is best to stick with `sys.stdout` or `sys.stdout.buffer`.  I couldn't get the built-in SDK Tap tests to work until I changed back to using `sys.stdout.buffer` in a tap utilizing msgspec.  I borrowed your use of `io.TextIOWrapper(buffer=io.BytesIO())` in the test_msgspec.py file that is part of this PR and added it to `singer_sdk.testing.runners.TapTestRunner._execute_sync` so the `redirect_stdout` doesn't error because of `stdout_buf = io.StringIO()`  lack of a `.buffer` to write to.

```
    def _execute_sync(self) -> tuple[str, str]:
        """Invoke a Tap object and return STDOUT and STDERR results in StringIO buffers.

        Returns:
            A 2-item tuple with StringIO buffers from the Tap's output: (stdout, stderr)
        """
        stdout_buf = io.TextIOWrapper(buffer=io.BytesIO())
        stderr_buf = io.TextIOWrapper(buffer=io.BytesIO())
        with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
            self.run_sync_dry_run()
        stdout_buf.seek(0)
        stderr_buf.seek(0)
        return stdout_buf.read(), stderr_buf.read()
```

In summary I had to change the tap's overwrite of `write_message` to utilize `sys.stdout.buffer.write()` and `sys.stdout.flush()`.  Then changed `singer_sdk.testing.runners.TapTestRunner._execute_sync` to use `io.TextIOWrapper(buffer=io.BytesIO())` instead of `io.StringIO()` for `stdout_buff` and `stderr_buff`.  The Singer-SDK built in tap tests ran correctly after the change to the tap and installing a local development branch of the singer-sdk with the change to `TapTestRunner._execute_sync` in the virtual environment I ran the tap test from.
edgarrmondragon: Ok, the tests are passing.

Now I want to think of how to make it easy and straightforward for a developer to use msgspec as the SerDe layer, and also keep the door open to _the user_ being the one deciding which serialization layer to use.
edgarrmondragon: @sourcery-ai review
sourcery-ai[bot]: <!-- Generated by sourcery-ai[bot]: start review_guide -->

## Reviewer's Guide by Sourcery

This pull request refactors the SDK to use msgspec for encoding and decoding Singer messages, which should improve performance. It also updates the test suite to use the new encoding.

#### Sequence diagram for MsgSpec message encoding flow

```mermaid
sequenceDiagram
    participant T as Tap
    participant MW as MsgSpecWriter
    participant SO as System Output

    T->>MW: write_message(message)
    activate MW
    MW->>MW: serialize_message(message)
    MW->>MW: format_message()
    MW->>SO: write to stdout buffer
    MW->>SO: flush stdout
    deactivate MW
```

#### Class diagram for MsgSpec encoding implementation

```mermaid
classDiagram
    class GenericSingerReader~T~ {
        <<abstract>>
        +default_input
        +deserialize_json(line: str)* dict
    }
    class GenericSingerWriter~T, M~ {
        <<abstract>>
        +serialize_message(message: M)* T
        +write_message(message: M) void
    }
    class MsgSpecReader {
        +default_input: sys.stdin
        +deserialize_json(line: str) dict
    }
    class MsgSpecWriter {
        +serialize_message(message: Message) bytes
        +write_message(message: Message) void
    }
    class Message {
        +to_dict() dict
    }

    GenericSingerReader <|-- MsgSpecReader
    GenericSingerWriter <|-- MsgSpecWriter
    MsgSpecWriter ..> Message

    note for MsgSpecReader "Uses msgspec for fast JSON deserialization"
    note for MsgSpecWriter "Uses msgspec for fast JSON serialization"
```

### File-Level Changes

| Change | Details | Files |
| ------ | ------- | ----- |
| Implemented msgspec encoding and decoding for Singer messages. | <ul><li>Added `msgspec` as a dependency.</li><li>Created `MsgSpecReader` and `MsgSpecWriter` classes.</li><li>Implemented `enc_hook` and `dec_hook` for handling non-native types.</li><li>Implemented `serialize_jsonl` for serializing messages.</li><li>Added tests for the new encoding.</li></ul> | `singer_sdk/_singerlib/encoding/_msgspec.py`<br/>`tests/_singerlib/encoding/test_msgspec.py` |
| Updated the test suite to use the new encoding. | <ul><li>Removed the old tests for the simple encoding.</li><li>Added benchmarks for the new encoding.</li><li>Updated the test suite to use the new encoding.</li><li>Updated the test suite to use TextIOWrapper instead of StringIO.</li></ul> | `tests/core/test_io.py`<br/>`tests/samples/test_tap_countries.py`<br/>`tests/_singerlib/encoding/test_simple.py` |
| Updated sample taps to use the new encoding. | <ul><li>Updated the sample taps to use the new `MsgSpecWriter` class.</li></ul> | `samples/sample_tap_countries/countries_tap.py`<br/>`samples/sample_tap_sqlite/__init__.py` |
| Updated noxfile to include msgspec. | <ul><li>Added `msgspec` to the list of extras in `noxfile.py`.</li><li>Added `msgspec` as a dependency in `pyproject.toml`.</li></ul> | `noxfile.py`<br/>`pyproject.toml` |

---

<details>
<summary>Tips and commands</summary>

#### Interacting with Sourcery

- **Trigger a new review:** Comment `@sourcery-ai review` on the pull request.
- **Continue discussions:** Reply directly to Sourcery's review comments.
- **Generate a GitHub issue from a review comment:** Ask Sourcery to create an
  issue from a review comment by replying to it.
- **Generate a pull request title:** Write `@sourcery-ai` anywhere in the pull
  request title to generate a title at any time.
- **Generate a pull request summary:** Write `@sourcery-ai summary` anywhere in
  the pull request body to generate a PR summary at any time. You can also use
  this command to specify where the summary should be inserted.

#### Customizing Your Experience

Access your [dashboard](https://app.sourcery.ai) to:
- Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
- Change the review language.
- Add, remove or edit custom review instructions.
- Adjust other review settings.

#### Getting Help

- [Contact our support team](mailto:support@sourcery.ai) for questions or feedback.
- Visit our [documentation](https://docs.sourcery.ai) for detailed guides and information.
- Keep in touch with the Sourcery team by following us on [X/Twitter](https://x.com/SourceryAI), [LinkedIn](https://www.linkedin.com/company/sourcery-ai/) or [GitHub](https://github.com/sourcery-ai).

</details>

<!-- Generated by sourcery-ai[bot]: end review_guide -->

Patch:
diff --git a/docs/dev_guide.md b/docs/dev_guide.md
index e8bc68fa..09665b9a 100644
--- a/docs/dev_guide.md
+++ b/docs/dev_guide.md
@@ -263,54 +263,4 @@ This is automatically included in the most recent version of the tap and target
 
 ### Testing performance
 
-We've had success using [`viztracer`](https://github.com/gaogaotiantian/viztracer) to create flame graphs for SDK-based packages and find if there are any serious performance bottlenecks.
-
-You can start doing the same in your package. Start by installing `viztracer`.
-
-````{tab} Poetry
-```console
-$ poetry add --group dev viztracer
-```
-````
-
-````{tab} uv
-```console
-$ uv add --group dev viztracer
-```
-````
-
-Then simply run your package's CLI as normal, preceded by the `viztracer` command
-
-````{tab} Poetry
-```console
-$ poetry run viztracer my-tap
-$ poetry run viztracer -- my-target --config=config.json --input=messages.json
-```
-````
-
-````{tab} uv
-```console
-$ uv run viztracer my-tap
-$ uv run viztracer -- my-target --config=config.json --input=messages.json
-```
-````
-
-That command will produce a `result.json` file which you can explore with the `vizviewer` tool.
-
-````{tab} Poetry
-```console
-$ poetry run vizviewer result.json
-```
-````
-
-````{tab} uv
-```console
-$ uv run vizviewer result.json
-```
-````
-
-The output should look like this
-
-![SDK Flame Graph](https://gitlab.com/meltano/sdk/uploads/07633ba1217de6eb1bb0e018133c608d/_write_record_message.png)
-
-**Note**: Chrome seems to work best for running the `vizviewer` app.
+Moved to a separate page: [Measuring Performance](./guides/performance.md#measuring-performance).
diff --git a/docs/guides/index.md b/docs/guides/index.md
index 1c3a0e5a..89e3a0c1 100644
--- a/docs/guides/index.md
+++ b/docs/guides/index.md
@@ -9,6 +9,7 @@ porting
 pagination-classes
 custom-clis
 config-schema
+performance
 sql-tap
 sql-target
 
diff --git a/docs/guides/performance.md b/docs/guides/performance.md
new file mode 100644
index 00000000..6fe39d33
--- /dev/null
+++ b/docs/guides/performance.md
@@ -0,0 +1,85 @@
+# Tap and target performance
+
+The following sections provide different strategies to improve the performance of your tap or target.
+
+## Use `BATCH` mode
+
+See the [BATCH messages](/batch.md) page for more information.
+
+## Use a different message writer or reader
+
+Starting from version `0.45.0`, the Meltano Singer SDK supports different message writers and readers.
+
+For example, to update your tap to use a message writer based on the [msgspec](https://github.com/jcrist/msgspec/) serialization library:
+
+```python
+from singer_sdk.contrib.msgspec import MsgSpecWriter
+
+
+class MyTap(Tap):
+    message_writer_class = MsgSpecWriter
+```
+
+To update your target:
+
+```python
+from singer_sdk.contrib.msgspec import MsgSpecReader
+
+
+class MyTarget(Target):
+    message_reader_class = MsgSpecReader
+```
+
+## Measuring performance
+
+We've had success using [`viztracer`](https://github.com/gaogaotiantian/viztracer) to create flame graphs for SDK-based packages and find if there are any serious performance bottlenecks.
+
+You can start doing the same in your package. Start by installing `viztracer`.
+
+````{tab} Poetry
+```console
+$ poetry add --group dev viztracer
+```
+````
+
+````{tab} uv
+```console
+$ uv add --group dev viztracer
+```
+````
+
+Then simply run your package's CLI as normal, preceded by the `viztracer` command
+
+````{tab} Poetry
+```console
+$ poetry run viztracer my-tap
+$ poetry run viztracer -- my-target --config=config.json --input=messages.json
+```
+````
+
+````{tab} uv
+```console
+$ uv run viztracer my-tap
+$ uv run viztracer -- my-target --config=config.json --input=messages.json
+```
+````
+
+That command will produce a `result.json` file which you can explore with the `vizviewer` tool.
+
+````{tab} Poetry
+```console
+$ poetry run vizviewer result.json
+```
+````
+
+````{tab} uv
+```console
+$ uv run vizviewer result.json
+```
+````
+
+The output should look like this
+
+![SDK Flame Graph](https://gitlab.com/meltano/sdk/uploads/07633ba1217de6eb1bb0e018133c608d/_write_record_message.png)
+
+**Note**: Chrome seems to work best for running the `vizviewer` app.
diff --git a/noxfile.py b/noxfile.py
index 06689b64..41f8649c 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -45,7 +45,7 @@ def mypy(session: nox.Session) -> None:
     """Check types with mypy."""
     args = session.posargs or ["singer_sdk"]
     session.install(
-        ".[faker,jwt,parquet,s3,testing]",
+        ".[faker,jwt,msgspec,parquet,s3,testing]",
         "-c",
         "requirements/requirements.txt",
     )
@@ -61,6 +61,7 @@ def tests(session: nox.Session) -> None:
     extras = [
         "faker",
         "jwt",
+        "msgspec",
         "parquet",
         "s3",
     ]
@@ -90,7 +91,7 @@ def tests(session: nox.Session) -> None:
 @nox.session(python=main_python_version)
 def benches(session: nox.Session) -> None:
     """Run benchmarks."""
-    session.install(".[jwt,s3]", "-c", "requirements/requirements.txt")
+    session.install(".[jwt,msgspec,s3]", "-c", "requirements/requirements.txt")
     session.install("-r", "requirements/requirements.test.txt")
     session.run(
         "pytest",
@@ -104,7 +105,7 @@ def benches(session: nox.Session) -> None:
 def dependencies(session: nox.Session) -> None:
     """Check issues with dependencies."""
     session.install(
-        ".[faker,jwt,parquet,s3,ssh,testing]",
+        ".[faker,jwt,msgspec,parquet,s3,ssh,testing]",
         "-c",
         "requirements/requirements.txt",
     )
@@ -117,7 +118,11 @@ def update_snapshots(session: nox.Session) -> None:
     """Update pytest snapshots."""
     args = session.posargs or ["-m", "snapshot"]
 
-    session.install(".[faker,jwt,parquet]", "-c", "requirements/requirements.txt")
+    session.install(
+        ".[faker,jwt,msgspec,parquet]",
+        "-c",
+        "requirements/requirements.txt",
+    )
     session.install("-r", "requirements/requirements.test.txt")
     session.run("pytest", "--snapshot-update", *args)
 
diff --git a/pyproject.toml b/pyproject.toml
index 79890a3b..47ab0614 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -91,6 +91,11 @@ ssh = [
     "paramiko>=3.3.0",
 ]
 
+# msgspec extras
+msgspec = [
+    "msgspec>=0.19.0",
+]
+
 [project.urls]
 Homepage = "https://sdk.meltano.com/en/latest/"
 Repository = "https://github.com/meltano/sdk"
diff --git a/requirements/requirements.codspeed.txt b/requirements/requirements.codspeed.txt
index 3bef39d1..e44976c4 100644
--- a/requirements/requirements.codspeed.txt
+++ b/requirements/requirements.codspeed.txt
@@ -47,6 +47,7 @@ markdown-it-py==3.0.0
 markupsafe==3.0.2
 mdurl==0.1.2
 moto==5.1.0
+msgspec==0.19.0
 multidict==6.1.0
 numpy==2.0.2 ; python_full_version < '3.10'
 numpy==2.2.3 ; python_full_version >= '3.10'
diff --git a/requirements/requirements.txt b/requirements/requirements.txt
index 83de7126..960e3ed2 100644
--- a/requirements/requirements.txt
+++ b/requirements/requirements.txt
@@ -523,6 +523,43 @@ jsonschema==4.23.0 \
 jsonschema-specifications==2024.10.1 \
     --hash=sha256:0f38b83639958ce1152d02a7f062902c41c8fd20d558b0c34344292d417ae272 \
     --hash=sha256:a09a0680616357d9a0ecf05c12ad234479f549239d0f5b55f3deea67475da9bf
+msgspec==0.19.0 \
+    --hash=sha256:00e87ecfa9795ee5214861eab8326b0e75475c2e68a384002aa135ea2a27d909 \
+    --hash=sha256:047cfa8675eb3bad68722cfe95c60e7afabf84d1bd8938979dd2b92e9e4a9551 \
+    --hash=sha256:0553bbc77662e5708fe66aa75e7bd3e4b0f209709c48b299afd791d711a93c36 \
+    --hash=sha256:067f0de1c33cfa0b6a8206562efdf6be5985b988b53dd244a8e06f993f27c8c0 \
+    --hash=sha256:0684573a821be3c749912acf5848cce78af4298345cb2d7a8b8948a0a5a27cfe \
+    --hash=sha256:0f5c043ace7962ef188746e83b99faaa9e3e699ab857ca3f367b309c8e2c6b12 \
+    --hash=sha256:15c1e86fff77184c20a2932cd9742bf33fe23125fa3fcf332df9ad2f7d483044 \
+    --hash=sha256:19746b50be214a54239aab822964f2ac81e38b0055cca94808359d779338c10e \
+    --hash=sha256:2719647625320b60e2d8af06b35f5b12d4f4d281db30a15a1df22adb2295f633 \
+    --hash=sha256:317050bc0f7739cb30d257ff09152ca309bf5a369854bbf1e57dffc310c1f20f \
+    --hash=sha256:3b5541b2b3294e5ffabe31a09d604e23a88533ace36ac288fa32a420aa38d229 \
+    --hash=sha256:3be5c02e1fee57b54130316a08fe40cca53af92999a302a6054cd451700ea7db \
+    --hash=sha256:3c4ec642689da44618f68c90855a10edbc6ac3ff7c1d94395446c65a776e712a \
+    --hash=sha256:43bbb237feab761b815ed9df43b266114203f53596f9b6e6f00ebd79d178cdf2 \
+    --hash=sha256:45c8fb410670b3b7eb884d44a75589377c341ec1392b778311acdbfa55187716 \
+    --hash=sha256:4cfc033c02c3e0aec52b71710d7f84cb3ca5eb407ab2ad23d75631153fdb1f12 \
+    --hash=sha256:5f0f65f29b45e2816d8bded36e6b837a4bf5fb60ec4bc3c625fa2c6da4124537 \
+    --hash=sha256:604037e7cd475345848116e89c553aa9a233259733ab51986ac924ab1b976f8e \
+    --hash=sha256:60ef4bdb0ec8e4ad62e5a1f95230c08efb1f64f32e6e8dd2ced685bcc73858b5 \
+    --hash=sha256:695b832d0091edd86eeb535cd39e45f3919f48d997685f7ac31acb15e0a2ed90 \
+    --hash=sha256:6c7adf191e4bd3be0e9231c3b6dc20cf1199ada2af523885efc2ed218eafd011 \
+    --hash=sha256:70eaef4934b87193a27d802534dc466778ad8d536e296ae2f9334e182ac27b6c \
+    --hash=sha256:757b501fa57e24896cf40a831442b19a864f56d253679f34f260dcb002524a6c \
+    --hash=sha256:82b2c42c1b9ebc89e822e7e13bbe9d17ede0c23c187469fdd9505afd5a481314 \
+    --hash=sha256:a5bc1472223a643f5ffb5bf46ccdede7f9795078194f14edd69e3aab7020d327 \
+    --hash=sha256:aa77046904db764b0462036bc63ef71f02b75b8f72e9c9dd4c447d6da1ed8f8e \
+    --hash=sha256:ac7f7c377c122b649f7545810c6cd1b47586e3aa3059126ce3516ac7ccc6a6a9 \
+    --hash=sha256:ca06aa08e39bf57e39a258e1996474f84d0dd8130d486c00bec26d797b8c5446 \
+    --hash=sha256:d8dd848ee7ca7c8153462557655570156c2be94e79acec3561cf379581343259 \
+    --hash=sha256:d911c442571605e17658ca2b416fd8579c5050ac9adc5e00c2cb3126c97f73bc \
+    --hash=sha256:e695dad6897896e9384cf5e2687d9ae9feaef50e802f93602d35458e20d1fb19 \
+    --hash=sha256:e78f46ff39a427e10b4a61614a2777ad69559cc8d603a7c05681f5a595ea98f7 \
+    --hash=sha256:f04cad4385e20be7c7176bb8ae3dca54a08e9756cfc97bcdb4f18560c3042063 \
+    --hash=sha256:f12d30dd6266557aaaf0aa0f9580a9a8fbeadfa83699c487713e355ec5f0bd86 \
+    --hash=sha256:f98bd8962ad549c27d63845b50af3f53ec468b6318400c9f1adfe8b092d7b62f \
+    --hash=sha256:fe2c4bf29bf4e89790b3117470dea2c20b59932772483082c468b990d45fb947
 multidict==6.1.0 \
     --hash=sha256:052e10d2d37810b99cc170b785945421141bf7bb7d2f8799d431e7db229c385f \
     --hash=sha256:06809f4f0f7ab7ea2cabf9caca7d79c22c0758b58a71f9d32943ae13c7ace056 \
diff --git a/samples/sample_tap_countries/countries_tap.py b/samples/sample_tap_countries/countries_tap.py
index 0c390616..3253fd6b 100644
--- a/samples/sample_tap_countries/countries_tap.py
+++ b/samples/sample_tap_countries/countries_tap.py
@@ -13,6 +13,7 @@ from samples.sample_tap_countries.countries_streams import (
     CountriesStream,
 )
 from singer_sdk import Stream, Tap
+from singer_sdk.contrib.msgspec import MsgSpecWriter
 from singer_sdk.typing import PropertiesList
 
 
@@ -22,6 +23,8 @@ class SampleTapCountries(Tap):
     name: str = "sample-tap-countries"
     config_jsonschema = PropertiesList().to_dict()
 
+    message_writer_class = MsgSpecWriter
+
     def discover_streams(self) -> list[Stream]:
         """Return a list of discovered streams."""
         return [
diff --git a/samples/sample_tap_sqlite/__init__.py b/samples/sample_tap_sqlite/__init__.py
index 2215cbfe..0dd005db 100644
--- a/samples/sample_tap_sqlite/__init__.py
+++ b/samples/sample_tap_sqlite/__init__.py
@@ -6,6 +6,7 @@ import typing as t
 
 from singer_sdk import SQLConnector, SQLStream, SQLTap
 from singer_sdk import typing as th
+from singer_sdk.contrib.msgspec import MsgSpecWriter
 
 DB_PATH_CONFIG = "path_to_db"
 
@@ -42,6 +43,8 @@ class SQLiteStream(SQLStream):
 class SQLiteTap(SQLTap):
     """The Tap class for SQLite."""
 
+    message_writer_class = MsgSpecWriter
+
     name = "tap-sqlite-sample"
     default_stream_class = SQLiteStream
     config_jsonschema = th.PropertiesList(
diff --git a/samples/sample_target_sqlite/__init__.py b/samples/sample_target_sqlite/__init__.py
index 7274f51f..dcd57484 100644
--- a/samples/sample_target_sqlite/__init__.py
+++ b/samples/sample_target_sqlite/__init__.py
@@ -8,6 +8,7 @@ import typing as t
 
 from singer_sdk import SQLConnector, SQLSink, SQLTarget
 from singer_sdk import typing as th
+from singer_sdk.contrib.msgspec import MsgSpecReader
 
 DB_PATH_CONFIG = "path_to_db"
 
@@ -85,6 +86,8 @@ class SQLiteSink(SQLSink[SQLiteConnector]):
 class SQLiteTarget(SQLTarget):
     """The Tap class for SQLite."""
 
+    message_reader_class = MsgSpecReader
+
     name = "target-sqlite-sample"
     default_sink_class = SQLiteSink
     max_parallelism = 1
diff --git a/singer_sdk/contrib/msgspec.py b/singer_sdk/contrib/msgspec.py
new file mode 100644
index 00000000..5537a91d
--- /dev/null
+++ b/singer_sdk/contrib/msgspec.py
@@ -0,0 +1,111 @@
+"""Singer message encoding using msgspec."""
+
+from __future__ import annotations
+
+import datetime
+import decimal
+import logging
+import sys
+import typing as t
+
+import msgspec
+
+from singer_sdk.singerlib.encoding.base import GenericSingerReader, GenericSingerWriter
+from singer_sdk.singerlib.encoding.simple import Message
+from singer_sdk.singerlib.exceptions import InvalidInputLine
+
+logger = logging.getLogger(__name__)
+
+
+def enc_hook(obj: t.Any) -> t.Any:  # noqa: ANN401
+    """Encoding type helper for non native types.
+
+    Args:
+        obj: the item to be encoded
+
+    Returns:
+        The object converted to the appropriate type, default is str
+    """
+    return obj.isoformat(sep="T") if isinstance(obj, datetime.datetime) else str(obj)
+
+
+def dec_hook(type: type, obj: t.Any) -> t.Any:  # noqa: ARG001, A002, ANN401
+    """Decoding type helper for non native types.
+
+    Args:
+        type: the type given
+        obj: the item to be decoded
+
+    Returns:
+        The object converted to the appropriate type, default is str.
+    """
+    return str(obj)
+
+
+encoder = msgspec.json.Encoder(enc_hook=enc_hook, decimal_format="number")
+decoder = msgspec.json.Decoder(dec_hook=dec_hook, float_hook=decimal.Decimal)
+_jsonl_msg_buffer = bytearray(64)
+
+
+def serialize_jsonl(obj: object, **kwargs: t.Any) -> bytes:  # noqa: ARG001
+    """Serialize a dictionary into a line of jsonl.
+
+    Args:
+        obj: A Python object usually a dict.
+        **kwargs: Optional key word arguments.
+
+    Returns:
+        A bytes of serialized json.
+    """
+    encoder.encode_into(obj, _jsonl_msg_buffer)
+    _jsonl_msg_buffer.extend(b"\n")
+    return _jsonl_msg_buffer
+
+
+class MsgSpecReader(GenericSingerReader[str]):
+    """Base class for all plugins reading Singer messages as strings from stdin."""
+
+    default_input = sys.stdin
+
+    def deserialize_json(self, line: str) -> dict:  # noqa: PLR6301
+        """Deserialize a line of json.
+
+        Args:
+            line: A single line of json.
+
+        Returns:
+            A dictionary of the deserialized json.
+
+        Raises:
+            InvalidInputLine: If the line cannot be parsed
+        """
+        try:
+            return decoder.decode(line)  # type: ignore[no-any-return]
+        except msgspec.DecodeError as exc:
+            logger.exception("Unable to parse:\n%s", line)
+            msg = f"Unable to parse line as JSON: {line}"
+            raise InvalidInputLine(msg) from exc
+
+
+class MsgSpecWriter(GenericSingerWriter[bytes, Message]):
+    """Interface for all plugins writing Singer messages to stdout."""
+
+    def serialize_message(self, message: Message) -> bytes:  # noqa: PLR6301
+        """Serialize a dictionary into a line of json.
+
+        Args:
+            message: A Singer message object.
+
+        Returns:
+            A string of serialized json.
+        """
+        return serialize_jsonl(message.to_dict())
+
+    def write_message(self, message: Message) -> None:
+        """Write a message to stdout.
+
+        Args:
+            message: The message to write.
+        """
+        sys.stdout.buffer.write(self.format_message(message))
+        sys.stdout.flush()
diff --git a/singer_sdk/mapper_base.py b/singer_sdk/mapper_base.py
index e7812c53..6763b751 100644
--- a/singer_sdk/mapper_base.py
+++ b/singer_sdk/mapper_base.py
@@ -9,16 +9,39 @@ import click
 
 from singer_sdk.helpers._classproperty import classproperty
 from singer_sdk.helpers.capabilities import CapabilitiesEnum, PluginCapabilities
-from singer_sdk.io_base import SingerReader, SingerWriter
-from singer_sdk.plugin_base import PluginBase
+from singer_sdk.plugin_base import BaseSingerReader, BaseSingerWriter
 
 if t.TYPE_CHECKING:
+    from pathlib import PurePath
+
     import singer_sdk.singerlib as singer
+    from singer_sdk.singerlib.encoding.base import (
+        GenericSingerReader,
+        GenericSingerWriter,
+    )
 
 
-class InlineMapper(PluginBase, SingerReader, SingerWriter, metaclass=abc.ABCMeta):
+class InlineMapper(BaseSingerReader, BaseSingerWriter, metaclass=abc.ABCMeta):
     """Abstract base class for inline mappers."""
 
+    def __init__(
+        self,
+        *,
+        config: dict | PurePath | str | list[PurePath | str] | None = None,
+        parse_env_config: bool = False,
+        validate_config: bool = True,
+        message_reader: GenericSingerReader | None = None,
+        message_writer: GenericSingerWriter | None = None,
+    ) -> None:
+        """Initialize the inline mapper."""
+        super().__init__(
+            config=config,
+            parse_env_config=parse_env_config,
+            validate_config=validate_config,
+        )
+        self.message_reader = message_reader or self.message_reader_class()
+        self.message_writer = message_writer or self.message_writer_class()
+
     @classproperty
     def capabilities(self) -> list[CapabilitiesEnum]:  # noqa: PLR6301
         """Get capabilities.
diff --git a/singer_sdk/plugin_base.py b/singer_sdk/plugin_base.py
index e9017add..d3375f7f 100644
--- a/singer_sdk/plugin_base.py
+++ b/singer_sdk/plugin_base.py
@@ -32,6 +32,7 @@ from singer_sdk.helpers.capabilities import (
     CapabilitiesEnum,
     PluginCapabilities,
 )
+from singer_sdk.io_base import SingerMessageType, SingerReader, SingerWriter
 from singer_sdk.mapper import PluginMapper
 from singer_sdk.typing import (
     DEFAULT_JSONSCHEMA_VALIDATOR,
@@ -46,6 +47,11 @@ else:
 if t.TYPE_CHECKING:
     from jsonschema import ValidationError
 
+    from singer_sdk.singerlib.encoding.base import (
+        GenericSingerReader,
+        GenericSingerWriter,
+    )
+
 SDK_PACKAGE_NAME = "singer_sdk"
 
 JSONSchemaValidator = extend_validator_with_defaults(DEFAULT_JSONSCHEMA_VALIDATOR)
@@ -645,3 +651,187 @@ class PluginBase(metaclass=abc.ABCMeta):  # noqa: PLR0904
             A callable CLI object.
         """
         return cls.get_singer_command()
+
+
+_T = t.TypeVar("_T")
+
+
+class BaseSingerIO(PluginBase):
+    """Base class for Singer taps and targets."""
+
+    def __init__(
+        self,
+        *,
+        config: dict | PurePath | str | list[PurePath | str] | None = None,
+        parse_env_config: bool = False,
+        validate_config: bool = True,
+    ):
+        """Initialize the Singer tap or target.
+
+        Args:
+            config: May be one or more paths, either as str or PurePath objects, or
+                it can be a predetermined config dict.
+            parse_env_config: True to parse settings from env vars.
+            validate_config: True to require validation of config settings.
+        """
+        super().__init__(
+            config=config,
+            parse_env_config=parse_env_config,
+            validate_config=validate_config,
+        )
+
+    @classmethod
+    def instantiate_processor(cls, instance: _T | None, default_cls: type[_T]) -> _T:
+        """Instantiate a processor instance.
+
+        Args:
+            instance: The instance to instantiate.
+            default_cls: The default class to instantiate.
+
+        Returns:
+            The instantiated processor instance.
+        """
+        return instance or default_cls()
+
+
+class BaseSingerReader(BaseSingerIO, metaclass=abc.ABCMeta):
+    """Base class for Singer readers."""
+
+    message_reader_class: type[GenericSingerReader] = SingerReader
+    """The message writer class to use for writing messages."""
+
+    def __init__(
+        self,
+        *,
+        config: dict | PurePath | str | list[PurePath | str] | None = None,
+        parse_env_config: bool = False,
+        validate_config: bool = True,
+        message_reader: GenericSingerReader | None = None,
+    ) -> None:
+        """Initialize the Singer reader.
+
+        Args:
+            config: May be one or more paths, either as str or PurePath objects, or
+                it can be a predetermined config dict.
+            parse_env_config: True to parse settings from env vars.
+            validate_config: True to require validation of config settings.
+            message_reader: A message reader object.
+        """
+        super().__init__(
+            config=config,
+            parse_env_config=parse_env_config,
+            validate_config=validate_config,
+        )
+        self.message_reader = self.instantiate_processor(
+            message_reader,
+            self.message_reader_class,
+        )
+
+    @t.final
+    def listen(self, file_input: t.IO[str] | None = None) -> None:
+        """Read from input until all messages are processed.
+
+        Args:
+            file_input: Readable stream of messages. Defaults to standard in.
+        """
+        counter = self.process_lines(file_input)
+        line_count = sum(counter.values())
+
+        self.logger.info(
+            "Target '%s' completed reading %d lines of input "
+            "(%d schemas, %d records, %d batch manifests, %d state messages).",
+            self.name,
+            line_count,
+            counter[SingerMessageType.SCHEMA],
+            counter[SingerMessageType.RECORD],
+            counter[SingerMessageType.BATCH],
+            counter[SingerMessageType.STATE],
+        )
+        self.process_endofpipe()
+
+    @t.final
+    def process_lines(self, file_input: t.IO[str] | None) -> t.Counter[str]:
+        """Internal method to process jsonl lines from a Singer tap.
+
+        Args:
+            file_input: Readable stream of messages, each on a separate line.
+
+        Returns:
+            A counter object for the processed lines.
+        """
+        return self.message_reader.process_lines(
+            file_input,
+            callbacks={
+                SingerMessageType.SCHEMA: self._process_schema_message,
+                SingerMessageType.RECORD: self._process_record_message,
+                SingerMessageType.STATE: self._process_state_message,
+                SingerMessageType.ACTIVATE_VERSION: self._process_activate_version_message,  # noqa: E501
+                SingerMessageType.BATCH: self._process_batch_message,
+            },
+        )
+
+    def process_endofpipe(self) -> None:
+        """Process end of pipe."""
+
+    def _assert_line_requires(self, message_dict: dict, requires: set[str]) -> None:
+        """Assert that a message contains required fields.
+
+        Args:
+            message_dict: The message to check.
+            requires: The required fields.
+        """
+        self.message_reader.assert_line_requires(message_dict, requires)
+
+    @abc.abstractmethod
+    def _process_schema_message(self, message_dict: dict) -> None: ...
+
+    @abc.abstractmethod
+    def _process_record_message(self, message_dict: dict) -> None: ...
+
+    @abc.abstractmethod
+    def _process_state_message(self, message_dict: dict) -> None: ...
+
+    @abc.abstractmethod
+    def _process_activate_version_message(self, message_dict: dict) -> None: ...
+
+    @abc.abstractmethod
+    def _process_batch_message(self, message_dict: dict) -> None: ...
+
+
+class BaseSingerWriter(BaseSingerIO):
+    """Base class for Singer writers."""
+
+    message_writer_class: type[GenericSingerWriter] = SingerWriter
+    """The message writer class to use for writing messages."""
+
+    def __init__(
+        self,
+        *,
+        config: dict | PurePath | str | list[PurePath | str] | None = None,
+        parse_env_config: bool = False,
+        validate_config: bool = True,
+        message_writer: GenericSingerWriter | None = None,
+    ) -> None:
+        """Initialize the Singer writer.
+
+        Args:
+            config: May be one or more paths, either as str or PurePath objects, or
+                it can be a predetermined config dict.
+            parse_env_config: True to parse settings from env vars.
+            validate_config: True to require validation of config settings.
+            message_writer: A message writer object.
+        """
+        super().__init__(
+            config=config,
+            parse_env_config=parse_env_config,
+            validate_config=validate_config,
+        )
+        self.message_writer = self.instantiate_processor(
+            message_writer,
+            self.message_writer_class,
+        )
+
+    @t.final
+    def write_message(self, message: t.Any) -> None:  # noqa: ANN401
+        """Write a message to the tap's message writer."""
+        self.message_writer.write_message(message)
diff --git a/singer_sdk/singerlib/encoding/base.py b/singer_sdk/singerlib/encoding/base.py
index 017e5296..69a6a507 100644
--- a/singer_sdk/singerlib/encoding/base.py
+++ b/singer_sdk/singerlib/encoding/base.py
@@ -43,55 +43,31 @@ class GenericSingerReader(t.Generic[T], metaclass=abc.ABCMeta):
         super().__init__()
         self._current_message: T | None = None
 
-    @t.final
-    def listen(self, file_input: t.IO[T] | None = None) -> None:
-        """Read from input until all messages are processed.
-
-        Args:
-            file_input: Readable stream of messages. Defaults to standard in.
-        """
-        try:
-            self._process_lines(file_input or self.default_input)
-        except Exception:
-            logger.debug(
-                "Failed while processing Singer message: %s",
-                self._current_message,
-            )
-            raise
-        self._process_endofpipe()
-
-    def _process_lines(self, file_input: t.IO[T]) -> t.Counter[str]:
+    def process_lines(
+        self,
+        file_input: t.IO[T] | None,
+        callbacks: dict[str, t.Callable[[dict], None]],
+    ) -> t.Counter[str]:
         """Internal method to process jsonl lines from a Singer tap.
 
         Args:
             file_input: Readable stream of messages, each on a separate line.
+            callbacks: Dictionary of message type to callback function.
 
         Returns:
             A counter object for the processed lines.
         """
         stats: dict[str, int] = defaultdict(int)
-        for line in file_input:
+        filein = file_input or self.default_input
+        for line in filein:
             self._current_message = line
 
             line_dict = self.deserialize_json(line)
-            self._assert_line_requires(line_dict, requires={"type"})
+            self.assert_line_requires(line_dict, requires={"type"})
 
             record_type: SingerMessageType = line_dict["type"]
-            if record_type == SingerMessageType.SCHEMA:
-                self._process_schema_message(line_dict)
-
-            elif record_type == SingerMessageType.RECORD:
-                self._process_record_message(line_dict)
-
-            elif record_type == SingerMessageType.ACTIVATE_VERSION:
-                self._process_activate_version_message(line_dict)
-
-            elif record_type == SingerMessageType.STATE:
-                self._process_state_message(line_dict)
-
-            elif record_type == SingerMessageType.BATCH:
-                self._process_batch_message(line_dict)
-
+            if callback := callbacks.get(record_type):
+                callback(line_dict)
             else:
                 self._process_unknown_message(line_dict)
 
@@ -105,7 +81,7 @@ class GenericSingerReader(t.Generic[T], metaclass=abc.ABCMeta):
         """Default input stream for the reader."""
 
     @staticmethod
-    def _assert_line_requires(line_dict: dict, requires: set[str]) -> None:
+    def assert_line_requires(line_dict: dict, requires: set[str]) -> None:
         """Check if dictionary .
 
         Args:
@@ -124,21 +100,6 @@ class GenericSingerReader(t.Generic[T], metaclass=abc.ABCMeta):
     def deserialize_json(self, line: T) -> dict:
         """Deserialize a line of json."""
 
-    @abc.abstractmethod
-    def _process_schema_message(self, message_dict: dict) -> None: ...
-
-    @abc.abstractmethod
-    def _process_record_message(self, message_dict: dict) -> None: ...
-
-    @abc.abstractmethod
-    def _process_state_message(self, message_dict: dict) -> None: ...
-
-    @abc.abstractmethod
-    def _process_activate_version_message(self, message_dict: dict) -> None: ...
-
-    @abc.abstractmethod
-    def _process_batch_message(self, message_dict: dict) -> None: ...
-
     def _process_unknown_message(self, message_dict: dict) -> None:  # noqa: PLR6301
         """Internal method to process unknown message types from a Singer tap.
 
@@ -152,9 +113,6 @@ class GenericSingerReader(t.Generic[T], metaclass=abc.ABCMeta):
         msg = f"Unknown message type '{record_type}' in message."
         raise ValueError(msg)
 
-    def _process_endofpipe(self) -> None:  # noqa: PLR6301
-        logger.debug("End of pipe reached")
-
 
 class GenericSingerWriter(t.Generic[T, M], metaclass=abc.ABCMeta):
     """Interface for all plugins writing Singer messages as strings or bytes."""
diff --git a/singer_sdk/tap_base.py b/singer_sdk/tap_base.py
index b572ad85..c231d0b4 100644
--- a/singer_sdk/tap_base.py
+++ b/singer_sdk/tap_base.py
@@ -30,7 +30,7 @@ from singer_sdk.helpers.capabilities import (
     TapCapabilities,
 )
 from singer_sdk.io_base import SingerWriter
-from singer_sdk.plugin_base import PluginBase
+from singer_sdk.plugin_base import BaseSingerWriter, PluginBase
 from singer_sdk.singerlib import Catalog, StateMessage
 
 if t.TYPE_CHECKING:
@@ -38,6 +38,7 @@ if t.TYPE_CHECKING:
 
     from singer_sdk.connectors import SQLConnector
     from singer_sdk.mapper import PluginMapper
+    from singer_sdk.singerlib.encoding.base import GenericSingerWriter
     from singer_sdk.streams import SQLStream, Stream
 
 STREAM_MAPS_CONFIG = "stream_maps"
@@ -51,7 +52,7 @@ class CliTestOptionValue(Enum):
     Disabled = "disabled"
 
 
-class Tap(PluginBase, SingerWriter, metaclass=abc.ABCMeta):  # noqa: PLR0904
+class Tap(BaseSingerWriter, metaclass=abc.ABCMeta):  # noqa: PLR0904
     """Abstract base class for taps.
 
     The Tap class governs configuration, validation, and stream discovery for tap
@@ -62,6 +63,9 @@ class Tap(PluginBase, SingerWriter, metaclass=abc.ABCMeta):  # noqa: PLR0904
     """Whether the tap's catalog is dynamic. Set to True if the catalog is
     generated dynamically (e.g. by querying a database's system tables)."""
 
+    message_writer_class: type[GenericSingerWriter] = SingerWriter
+    """The message writer class to use for writing messages."""
+
     # Constructor
 
     def __init__(
@@ -73,6 +77,7 @@ class Tap(PluginBase, SingerWriter, metaclass=abc.ABCMeta):  # noqa: PLR0904
         parse_env_config: bool = False,
         validate_config: bool = True,
         setup_mapper: bool = True,
+        message_writer: GenericSingerWriter | None = None,
     ) -> None:
         """Initialize the tap.
 
@@ -86,11 +91,13 @@ class Tap(PluginBase, SingerWriter, metaclass=abc.ABCMeta):  # noqa: PLR0904
                 variables.
             validate_config: True to require validation of config settings.
             setup_mapper: True to initialize the plugin mapper.
+            message_writer: The class class to use for writing Singer messages.
         """
         super().__init__(
             config=config,
             parse_env_config=parse_env_config,
             validate_config=validate_config,
+            message_writer=message_writer,
         )
 
         # Declare private members
diff --git a/singer_sdk/target_base.py b/singer_sdk/target_base.py
index c30c2084..241c0b32 100644
--- a/singer_sdk/target_base.py
+++ b/singer_sdk/target_base.py
@@ -28,20 +28,21 @@ from singer_sdk.helpers.capabilities import (
     PluginCapabilities,
     TargetCapabilities,
 )
-from singer_sdk.io_base import SingerMessageType, SingerReader
-from singer_sdk.plugin_base import PluginBase
+from singer_sdk.io_base import SingerReader
+from singer_sdk.plugin_base import BaseSingerReader
 
 if t.TYPE_CHECKING:
     from pathlib import PurePath
 
     from singer_sdk.connectors import SQLConnector
     from singer_sdk.mapper import PluginMapper
+    from singer_sdk.singerlib.encoding.base import GenericSingerReader
     from singer_sdk.sinks import Sink, SQLSink
 
 _MAX_PARALLELISM = 8
 
 
-class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
+class Target(BaseSingerReader, metaclass=abc.ABCMeta):
     """Abstract base class for targets.
 
     The `Target` class manages config information and is responsible for processing the
@@ -57,6 +58,9 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
     # Required if `Target.get_sink_class()` is not defined.
     default_sink_class: type[Sink]
 
+    message_reader_class: type[GenericSingerReader] = SingerReader
+    """The message reader class to use for reading messages."""
+
     def __init__(
         self,
         *,
@@ -64,6 +68,7 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
         parse_env_config: bool = False,
         validate_config: bool = True,
         setup_mapper: bool = True,
+        message_reader: GenericSingerReader | None = None,
     ) -> None:
         """Initialize the target.
 
@@ -75,11 +80,13 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
                 variables.
             validate_config: True to require validation of config settings.
             setup_mapper: True to setup the mapper. Set to False if you want to
+            message_reader: The message reader to use for reading messages.
         """
         super().__init__(
             config=config,
             parse_env_config=parse_env_config,
             validate_config=validate_config,
+            message_reader=message_reader,
         )
 
         self._latest_state: dict[str, dict] = {}
@@ -292,34 +299,7 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
             )
             self.drain_all()
 
-    def _process_lines(self, file_input: t.IO[str]) -> t.Counter[str]:
-        """Internal method to process jsonl lines from a Singer tap.
-
-        Args:
-            file_input: Readable stream of messages, each on a separate line.
-
-        Returns:
-            A counter object for the processed lines.
-        """
-        self.logger.info("Target '%s' is listening for input from tap.", self.name)
-        counter = super()._process_lines(file_input)
-
-        line_count = sum(counter.values())
-
-        self.logger.info(
-            "Target '%s' completed reading %d lines of input "
-            "(%d schemas, %d records, %d batch manifests, %d state messages).",
-            self.name,
-            line_count,
-            counter[SingerMessageType.SCHEMA],
-            counter[SingerMessageType.RECORD],
-            counter[SingerMessageType.BATCH],
-            counter[SingerMessageType.STATE],
-        )
-
-        return counter
-
-    def _process_endofpipe(self) -> None:
+    def process_endofpipe(self) -> None:
         """Called after all input lines have been read."""
         self.drain_all(is_endofpipe=True)
 
@@ -496,10 +476,8 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
         This method is internal to the SDK and should not need to be overridden.
 
         Args:
-            is_endofpipe: This is passed by the
-                          :meth:`~singer_sdk.Sink._process_endofpipe()` which
-                          is called after the target instance has finished
-                          listening to the stdin
+            is_endofpipe: This is called after the target instance has finished
+                listening to the stdin.
         """
         state = copy.deepcopy(self._latest_state)
         self._drain_all(self._sinks_to_clear, 1)
diff --git a/tests/_singerlib/test_catalog.py b/tests/singerlib/test_catalog.py
similarity index 100%
rename from tests/_singerlib/test_catalog.py
rename to tests/singerlib/test_catalog.py
diff --git a/tests/_singerlib/test_messages.py b/tests/singerlib/test_messages.py
similarity index 100%
rename from tests/_singerlib/test_messages.py
rename to tests/singerlib/test_messages.py
diff --git a/tests/_singerlib/test_schema.py b/tests/singerlib/test_schema.py
similarity index 100%
rename from tests/_singerlib/test_schema.py
rename to tests/singerlib/test_schema.py
diff --git a/tests/_singerlib/test_utils.py b/tests/singerlib/test_utils.py
similarity index 100%
rename from tests/_singerlib/test_utils.py
rename to tests/singerlib/test_utils.py
diff --git a/uv.lock b/uv.lock
index 4bba9fe2..a2f77c6c 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1229,6 +1229,49 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/91/d7/0dadf1e7ee7be7920dc0d8d3ee9e0fa9af13f2eedec7fffc4475a91697a6/moto-5.1.0-py3-none-any.whl", hash = "sha256:4fada00cedfba661aa58fe0b33b3ba9a0ef96d0e9937c9bed5163053898b4a27", size = 4725360 },
 ]
 
+[[package]]
+name = "msgspec"
+version = "0.19.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/cf/9b/95d8ce458462b8b71b8a70fa94563b2498b89933689f3a7b8911edfae3d7/msgspec-0.19.0.tar.gz", hash = "sha256:604037e7cd475345848116e89c553aa9a233259733ab51986ac924ab1b976f8e", size = 216934 }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/13/40/817282b42f58399762267b30deb8ac011d8db373f8da0c212c85fbe62b8f/msgspec-0.19.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:d8dd848ee7ca7c8153462557655570156c2be94e79acec3561cf379581343259", size = 190019 },
+    { url = "https://files.pythonhosted.org/packages/92/99/bd7ed738c00f223a8119928661167a89124140792af18af513e6519b0d54/msgspec-0.19.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:0553bbc77662e5708fe66aa75e7bd3e4b0f209709c48b299afd791d711a93c36", size = 183680 },
+    { url = "https://files.pythonhosted.org/packages/e5/27/322badde18eb234e36d4a14122b89edd4e2973cdbc3da61ca7edf40a1ccd/msgspec-0.19.0-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:fe2c4bf29bf4e89790b3117470dea2c20b59932772483082c468b990d45fb947", size = 209334 },
+    { url = "https://files.pythonhosted.org/packages/c6/65/080509c5774a1592b2779d902a70b5fe008532759927e011f068145a16cb/msgspec-0.19.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:00e87ecfa9795ee5214861eab8326b0e75475c2e68a384002aa135ea2a27d909", size = 211551 },
+    { url = "https://files.pythonhosted.org/packages/6f/2e/1c23c6b4ca6f4285c30a39def1054e2bee281389e4b681b5e3711bd5a8c9/msgspec-0.19.0-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:3c4ec642689da44618f68c90855a10edbc6ac3ff7c1d94395446c65a776e712a", size = 215099 },
+    { url = "https://files.pythonhosted.org/packages/83/fe/95f9654518879f3359d1e76bc41189113aa9102452170ab7c9a9a4ee52f6/msgspec-0.19.0-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:2719647625320b60e2d8af06b35f5b12d4f4d281db30a15a1df22adb2295f633", size = 218211 },
+    { url = "https://files.pythonhosted.org/packages/79/f6/71ca7e87a1fb34dfe5efea8156c9ef59dd55613aeda2ca562f122cd22012/msgspec-0.19.0-cp310-cp310-win_amd64.whl", hash = "sha256:695b832d0091edd86eeb535cd39e45f3919f48d997685f7ac31acb15e0a2ed90", size = 186174 },
+    { url = "https://files.pythonhosted.org/packages/24/d4/2ec2567ac30dab072cce3e91fb17803c52f0a37aab6b0c24375d2b20a581/msgspec-0.19.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:aa77046904db764b0462036bc63ef71f02b75b8f72e9c9dd4c447d6da1ed8f8e", size = 187939 },
+    { url = "https://files.pythonhosted.org/packages/2b/c0/18226e4328897f4f19875cb62bb9259fe47e901eade9d9376ab5f251a929/msgspec-0.19.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:047cfa8675eb3bad68722cfe95c60e7afabf84d1bd8938979dd2b92e9e4a9551", size = 182202 },
+    { url = "https://files.pythonhosted.org/packages/81/25/3a4b24d468203d8af90d1d351b77ea3cffb96b29492855cf83078f16bfe4/msgspec-0.19.0-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:e78f46ff39a427e10b4a61614a2777ad69559cc8d603a7c05681f5a595ea98f7", size = 209029 },
+    { url = "https://files.pythonhosted.org/packages/85/2e/db7e189b57901955239f7689b5dcd6ae9458637a9c66747326726c650523/msgspec-0.19.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6c7adf191e4bd3be0e9231c3b6dc20cf1199ada2af523885efc2ed218eafd011", size = 210682 },
+    { url = "https://files.pythonhosted.org/packages/03/97/7c8895c9074a97052d7e4a1cc1230b7b6e2ca2486714eb12c3f08bb9d284/msgspec-0.19.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:f04cad4385e20be7c7176bb8ae3dca54a08e9756cfc97bcdb4f18560c3042063", size = 214003 },
+    { url = "https://files.pythonhosted.org/packages/61/61/e892997bcaa289559b4d5869f066a8021b79f4bf8e955f831b095f47a4cd/msgspec-0.19.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:45c8fb410670b3b7eb884d44a75589377c341ec1392b778311acdbfa55187716", size = 216833 },
+    { url = "https://files.pythonhosted.org/packages/ce/3d/71b2dffd3a1c743ffe13296ff701ee503feaebc3f04d0e75613b6563c374/msgspec-0.19.0-cp311-cp311-win_amd64.whl", hash = "sha256:70eaef4934b87193a27d802534dc466778ad8d536e296ae2f9334e182ac27b6c", size = 186184 },
+    { url = "https://files.pythonhosted.org/packages/b2/5f/a70c24f075e3e7af2fae5414c7048b0e11389685b7f717bb55ba282a34a7/msgspec-0.19.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:f98bd8962ad549c27d63845b50af3f53ec468b6318400c9f1adfe8b092d7b62f", size = 190485 },
+    { url = "https://files.pythonhosted.org/packages/89/b0/1b9763938cfae12acf14b682fcf05c92855974d921a5a985ecc197d1c672/msgspec-0.19.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:43bbb237feab761b815ed9df43b266114203f53596f9b6e6f00ebd79d178cdf2", size = 183910 },
+    { url = "https://files.pythonhosted.org/packages/87/81/0c8c93f0b92c97e326b279795f9c5b956c5a97af28ca0fbb9fd86c83737a/msgspec-0.19.0-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:4cfc033c02c3e0aec52b71710d7f84cb3ca5eb407ab2ad23d75631153fdb1f12", size = 210633 },
+    { url = "https://files.pythonhosted.org/packages/d0/ef/c5422ce8af73928d194a6606f8ae36e93a52fd5e8df5abd366903a5ca8da/msgspec-0.19.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d911c442571605e17658ca2b416fd8579c5050ac9adc5e00c2cb3126c97f73bc", size = 213594 },
+    { url = "https://files.pythonhosted.org/packages/19/2b/4137bc2ed45660444842d042be2cf5b18aa06efd2cda107cff18253b9653/msgspec-0.19.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:757b501fa57e24896cf40a831442b19a864f56d253679f34f260dcb002524a6c", size = 214053 },
+    { url = "https://files.pythonhosted.org/packages/9d/e6/8ad51bdc806aac1dc501e8fe43f759f9ed7284043d722b53323ea421c360/msgspec-0.19.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:5f0f65f29b45e2816d8bded36e6b837a4bf5fb60ec4bc3c625fa2c6da4124537", size = 219081 },
+    { url = "https://files.pythonhosted.org/packages/b1/ef/27dd35a7049c9a4f4211c6cd6a8c9db0a50647546f003a5867827ec45391/msgspec-0.19.0-cp312-cp312-win_amd64.whl", hash = "sha256:067f0de1c33cfa0b6a8206562efdf6be5985b988b53dd244a8e06f993f27c8c0", size = 187467 },
+    { url = "https://files.pythonhosted.org/packages/3c/cb/2842c312bbe618d8fefc8b9cedce37f773cdc8fa453306546dba2c21fd98/msgspec-0.19.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:f12d30dd6266557aaaf0aa0f9580a9a8fbeadfa83699c487713e355ec5f0bd86", size = 190498 },
+    { url = "https://files.pythonhosted.org/packages/58/95/c40b01b93465e1a5f3b6c7d91b10fb574818163740cc3acbe722d1e0e7e4/msgspec-0.19.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:82b2c42c1b9ebc89e822e7e13bbe9d17ede0c23c187469fdd9505afd5a481314", size = 183950 },
+    { url = "https://files.pythonhosted.org/packages/e8/f0/5b764e066ce9aba4b70d1db8b087ea66098c7c27d59b9dd8a3532774d48f/msgspec-0.19.0-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:19746b50be214a54239aab822964f2ac81e38b0055cca94808359d779338c10e", size = 210647 },
+    { url = "https://files.pythonhosted.org/packages/9d/87/bc14f49bc95c4cb0dd0a8c56028a67c014ee7e6818ccdce74a4862af259b/msgspec-0.19.0-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:60ef4bdb0ec8e4ad62e5a1f95230c08efb1f64f32e6e8dd2ced685bcc73858b5", size = 213563 },
+    { url = "https://files.pythonhosted.org/packages/53/2f/2b1c2b056894fbaa975f68f81e3014bb447516a8b010f1bed3fb0e016ed7/msgspec-0.19.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:ac7f7c377c122b649f7545810c6cd1b47586e3aa3059126ce3516ac7ccc6a6a9", size = 213996 },
+    { url = "https://files.pythonhosted.org/packages/aa/5a/4cd408d90d1417e8d2ce6a22b98a6853c1b4d7cb7669153e4424d60087f6/msgspec-0.19.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:a5bc1472223a643f5ffb5bf46ccdede7f9795078194f14edd69e3aab7020d327", size = 219087 },
+    { url = "https://files.pythonhosted.org/packages/23/d8/f15b40611c2d5753d1abb0ca0da0c75348daf1252220e5dda2867bd81062/msgspec-0.19.0-cp313-cp313-win_amd64.whl", hash = "sha256:317050bc0f7739cb30d257ff09152ca309bf5a369854bbf1e57dffc310c1f20f", size = 187432 },
+    { url = "https://files.pythonhosted.org/packages/ea/d0/323f867eaec1f2236ba30adf613777b1c97a7e8698e2e881656b21871fa4/msgspec-0.19.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:15c1e86fff77184c20a2932cd9742bf33fe23125fa3fcf332df9ad2f7d483044", size = 189926 },
+    { url = "https://files.pythonhosted.org/packages/a8/37/c3e1b39bdae90a7258d77959f5f5e36ad44b40e2be91cff83eea33c54d43/msgspec-0.19.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:3b5541b2b3294e5ffabe31a09d604e23a88533ace36ac288fa32a420aa38d229", size = 183873 },
+    { url = "https://files.pythonhosted.org/packages/cb/a2/48f2c15c7644668e51f4dce99d5f709bd55314e47acb02e90682f5880f35/msgspec-0.19.0-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:0f5c043ace7962ef188746e83b99faaa9e3e699ab857ca3f367b309c8e2c6b12", size = 209272 },
+    { url = "https://files.pythonhosted.org/packages/25/3c/aa339cf08b990c3f07e67b229a3a8aa31bf129ed974b35e5daa0df7d9d56/msgspec-0.19.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ca06aa08e39bf57e39a258e1996474f84d0dd8130d486c00bec26d797b8c5446", size = 211396 },
+    { url = "https://files.pythonhosted.org/packages/c7/00/c7fb9d524327c558b2803973cc3f988c5100a1708879970a9e377bdf6f4f/msgspec-0.19.0-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:e695dad6897896e9384cf5e2687d9ae9feaef50e802f93602d35458e20d1fb19", size = 215002 },
+    { url = "https://files.pythonhosted.org/packages/3f/bf/d9f9fff026c1248cde84a5ce62b3742e8a63a3c4e811f99f00c8babf7615/msgspec-0.19.0-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:3be5c02e1fee57b54130316a08fe40cca53af92999a302a6054cd451700ea7db", size = 218132 },
+    { url = "https://files.pythonhosted.org/packages/00/03/b92011210f79794958167a3a3ea64a71135d9a2034cfb7597b545a42606d/msgspec-0.19.0-cp39-cp39-win_amd64.whl", hash = "sha256:0684573a821be3c749912acf5848cce78af4298345cb2d7a8b8948a0a5a27cfe", size = 186301 },
+]
+
 [[package]]
 name = "multidict"
 version = "6.1.0"
@@ -2341,6 +2384,9 @@ jwt = [
     { name = "cryptography" },
     { name = "pyjwt" },
 ]
+msgspec = [
+    { name = "msgspec" },
+]
 parquet = [
     { name = "numpy", version = "2.0.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" },
     { name = "numpy", version = "2.2.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" },
@@ -2467,6 +2513,7 @@ requires-dist = [
     { name = "joblib", specifier = ">=1.3.0" },
     { name = "jsonpath-ng", specifier = ">=1.5.3" },
     { name = "jsonschema", specifier = ">=4.16.0" },
+    { name = "msgspec", marker = "extra == 'msgspec'", specifier = ">=0.19.0" },
     { name = "numpy", marker = "extra == 'parquet'", specifier = ">=1.22" },
     { name = "packaging", specifier = ">=23.1" },
     { name = "paramiko", marker = "extra == 'ssh'", specifier = ">=3.3.0" },
@@ -2484,7 +2531,7 @@ requires-dist = [
     { name = "sqlalchemy", specifier = ">=1.4,<3.0" },
     { name = "typing-extensions", specifier = ">=4.5.0" },
 ]
-provides-extras = ["faker", "jwt", "parquet", "s3", "ssh", "testing"]
+provides-extras = ["faker", "jwt", "msgspec", "parquet", "s3", "ssh", "testing"]
 
 [package.metadata.requires-dev]
 benchmark = [


Test patch:
diff --git a/singer_sdk/testing/legacy.py b/singer_sdk/testing/legacy.py
index 89737d31..c39e0f1b 100644
--- a/singer_sdk/testing/legacy.py
+++ b/singer_sdk/testing/legacy.py
@@ -111,7 +111,12 @@ def get_standard_target_tests(
     return []
 
 
-def tap_sync_test(tap: Tap) -> tuple[io.StringIO, io.StringIO]:
+def tap_sync_test(
+    tap: Tap,
+) -> tuple[
+    io.TextIOWrapper[io.BytesIO],
+    io.TextIOWrapper[io.BytesIO],
+]:
     """Invokes a Tap object and return STDOUT and STDERR results in StringIO buffers.
 
     Args:
@@ -120,8 +125,8 @@ def tap_sync_test(tap: Tap) -> tuple[io.StringIO, io.StringIO]:
     Returns:
         A 2-item tuple with StringIO buffers from the Tap's output: (stdout, stderr)
     """
-    stdout_buf = io.StringIO()
-    stderr_buf = io.StringIO()
+    stdout_buf = io.TextIOWrapper(io.BytesIO(), encoding="utf-8")
+    stderr_buf = io.TextIOWrapper(io.BytesIO(), encoding="utf-8")
     with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
         tap.sync_all()
     stdout_buf.seek(0)
@@ -171,10 +176,10 @@ def _select_all(catalog_dict: dict) -> dict:
 
 def target_sync_test(
     target: Target,
-    input: io.StringIO | None,  # noqa: A002
+    input: t.IO[str] | None,  # noqa: A002
     *,
     finalize: bool = True,
-) -> tuple[io.StringIO, io.StringIO]:
+) -> tuple[io.TextIOWrapper[io.BytesIO], io.TextIOWrapper[io.BytesIO]]:
     """Invoke the target with the provided input.
 
     Args:
@@ -186,14 +191,14 @@ def target_sync_test(
     Returns:
         A 2-item tuple with StringIO buffers from the Target's output: (stdout, stderr)
     """
-    stdout_buf = io.StringIO()
-    stderr_buf = io.StringIO()
+    stdout_buf = io.TextIOWrapper(io.BytesIO(), encoding="utf-8")
+    stderr_buf = io.TextIOWrapper(io.BytesIO(), encoding="utf-8")
 
     with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
         if input is not None:
-            target._process_lines(input)  # noqa: SLF001
+            target.process_lines(input)
         if finalize:
-            target._process_endofpipe()  # noqa: SLF001
+            target.process_endofpipe()
 
     stdout_buf.seek(0)
     stderr_buf.seek(0)
@@ -203,7 +208,12 @@ def target_sync_test(
 def tap_to_target_sync_test(
     tap: Tap,
     target: Target,
-) -> tuple[io.StringIO, io.StringIO, io.StringIO, io.StringIO]:
+) -> tuple[
+    io.TextIOWrapper[io.BytesIO],
+    io.TextIOWrapper[io.BytesIO],
+    io.TextIOWrapper[io.BytesIO],
+    io.TextIOWrapper[io.BytesIO],
+]:
     """Test and end-to-end sink from the tap to the target.
 
     Note: This method buffers all output from the tap in memory and should not be
@@ -236,7 +246,7 @@ def sync_end_to_end(tap: Tap, target: Target, *mappers: InlineMapper) -> None:
         mappers: Zero or more inline mapper to apply in between the tap and target, in
             order.
     """
-    buf = io.StringIO()
+    buf = io.TextIOWrapper(io.BytesIO(), encoding="utf-8")
     with redirect_stdout(buf):
         tap.sync_all()
 
@@ -244,7 +254,7 @@ def sync_end_to_end(tap: Tap, target: Target, *mappers: InlineMapper) -> None:
     mapper_output = buf
 
     for mapper in mappers:
-        buf = io.StringIO()
+        buf = io.TextIOWrapper(io.BytesIO(), encoding="utf-8")
         with redirect_stdout(buf):
             mapper.listen(mapper_output)
 
diff --git a/singer_sdk/testing/runners.py b/singer_sdk/testing/runners.py
index b4ff913f..7e4217cb 100644
--- a/singer_sdk/testing/runners.py
+++ b/singer_sdk/testing/runners.py
@@ -185,8 +185,8 @@ class TapTestRunner(SingerTestRunner):
         Returns:
             A 2-item tuple with StringIO buffers from the Tap's output: (stdout, stderr)
         """
-        stdout_buf = io.StringIO()
-        stderr_buf = io.StringIO()
+        stdout_buf = io.TextIOWrapper(io.BytesIO(), encoding="utf-8")
+        stderr_buf = io.TextIOWrapper(io.BytesIO(), encoding="utf-8")
         with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
             self.run_sync_dry_run()
         stdout_buf.seek(0)
@@ -282,7 +282,7 @@ class TargetTestRunner(SingerTestRunner):
         target_input: t.IO[str],
         *,
         finalize: bool = True,
-    ) -> tuple[io.StringIO, io.StringIO]:
+    ) -> tuple[io.TextIOWrapper[io.BytesIO], io.TextIOWrapper[io.BytesIO]]:
         """Invoke the target with the provided input.
 
         Args:
@@ -295,14 +295,14 @@ class TargetTestRunner(SingerTestRunner):
             A 2-item tuple with StringIO buffers from the Target's output:
                 (stdout, stderr)
         """
-        stdout_buf = io.StringIO()
-        stderr_buf = io.StringIO()
+        stdout_buf = io.TextIOWrapper(io.BytesIO(), encoding="utf-8")
+        stderr_buf = io.TextIOWrapper(io.BytesIO(), encoding="utf-8")
 
         with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
             if target_input is not None:
-                target._process_lines(target_input)  # noqa: SLF001
+                target.process_lines(target_input)
             if finalize:
-                target._process_endofpipe()  # noqa: SLF001
+                target.process_endofpipe()
 
         stdout_buf.seek(0)
         stderr_buf.seek(0)
diff --git a/singer_sdk/testing/templates.py b/singer_sdk/testing/templates.py
index 73e154eb..9a7ede5c 100644
--- a/singer_sdk/testing/templates.py
+++ b/singer_sdk/testing/templates.py
@@ -6,6 +6,7 @@ import contextlib
 import importlib.resources
 import typing as t
 import warnings
+from functools import cached_property
 
 from singer_sdk.testing import target_test_streams
 from singer_sdk.testing.runners import SingerTestRunner, TapTestRunner, TargetTestRunner
@@ -230,7 +231,7 @@ class AttributeTestTemplate(TestTemplate[TapTestRunner]):
         self.attribute_name = attribute_name
         super().run(config, resource, runner)
 
-    @property
+    @cached_property
     def non_null_attribute_values(self) -> list[t.Any]:
         """Extract attribute values from stream records.
 
@@ -242,7 +243,10 @@ class AttributeTestTemplate(TestTemplate[TapTestRunner]):
             for r in self.stream_records
             if r.get(self.attribute_name) is not None
         ]
-        if not values:
+        if (
+            not values
+            and self.stream.name not in self.config.ignore_no_records_for_streams
+        ):
             warnings.warn(
                 UserWarning("No records were available to test."),
                 stacklevel=2,
diff --git a/tests/core/test_io.py b/tests/core/test_io.py
index 4f8b2019..eb79e085 100644
--- a/tests/core/test_io.py
+++ b/tests/core/test_io.py
@@ -2,132 +2,22 @@
 
 from __future__ import annotations
 
-import decimal
-import io
 import itertools
 import json
-import logging
-from contextlib import nullcontext, redirect_stdout
-from textwrap import dedent
 
 import pytest
 
 from singer_sdk._singerlib import RecordMessage
-from singer_sdk._singerlib.exceptions import InvalidInputLine
-from singer_sdk.io_base import SingerReader, SingerWriter
-
-
-class DummyReader(SingerReader):
-    def _process_activate_version_message(self, message_dict: dict) -> None:
-        pass
-
-    def _process_batch_message(self, message_dict: dict) -> None:
-        pass
-
-    def _process_record_message(self, message_dict: dict) -> None:
-        pass
-
-    def _process_schema_message(self, message_dict: dict) -> None:
-        pass
-
-    def _process_state_message(self, message_dict: dict) -> None:
-        pass
-
-
-@pytest.mark.parametrize(
-    "line,expected,exception",
-    [
-        pytest.param(
-            "not-valid-json",
-            None,
-            pytest.raises(InvalidInputLine),
-            id="unparsable",
-        ),
-        pytest.param(
-            '{"type": "RECORD", "stream": "users", "record": {"id": 1, "value": 1.23}}',
-            {
-                "type": "RECORD",
-                "stream": "users",
-                "record": {"id": 1, "value": decimal.Decimal("1.23")},
-            },
-            nullcontext(),
-            id="record",
-        ),
-    ],
-)
-def test_deserialize(line, expected, exception):
-    reader = DummyReader()
-    with exception:
-        assert reader.deserialize_json(line) == expected
-
-
-def test_listen():
-    reader = DummyReader()
-    input_lines = io.StringIO(
-        dedent("""\
-        {"type": "SCHEMA", "stream": "users", "schema": {"type": "object", "properties": {"id": {"type": "integer"}, "value": {"type": "number"}}}}
-        {"type": "RECORD", "stream": "users", "record": {"id": 1, "value": 1.23}}
-        {"type": "RECORD", "stream": "users", "record": {"id": 2, "value": 2.34}}
-        {"type": "STATE", "value": {"bookmarks": {"users": {"id": 2}}}}
-        {"type": "SCHEMA", "stream": "batches", "schema": {"type": "object", "properties": {"id": {"type": "integer"}, "value": {"type": "number"}}}}
-        {"type": "BATCH", "stream": "batches", "encoding": {"format": "jsonl", "compression": "gzip"}, "manifest": ["file1.jsonl.gz", "file2.jsonl.gz"]}
-        {"type": "STATE", "value": {"bookmarks": {"users": {"id": 2}, "batches": {"id": 1000000}}}}
-    """)  # noqa: E501
-    )
-    reader.listen(input_lines)
-
-
-def test_listen_unknown_message():
-    reader = DummyReader()
-    input_lines = io.StringIO('{"type": "UNKNOWN"}\n')
-    with pytest.raises(ValueError, match="Unknown message type"):
-        reader.listen(input_lines)
-
-
-def test_listen_error(caplog: pytest.LogCaptureFixture):
-    class ErrorReader(DummyReader):
-        def _process_record_message(self, message_dict: dict) -> None:  # noqa: ARG002
-            msg = "Bad record"
-            raise ValueError(msg)
-
-    message = RecordMessage(
-        stream="users",
-        record={"id": 1, "value": 1.23},
-    )
-
-    input_lines = io.StringIO(json.dumps(message.to_dict()) + "\n")
-
-    reader = ErrorReader()
-    with caplog.at_level(logging.DEBUG), pytest.raises(ValueError, match="Bad record"):
-        reader.listen(input_lines)
-
-    assert caplog.records
-
-    message = caplog.records[0].message
-    assert "Failed while processing Singer message" in message
-
-
-def test_write_message():
-    writer = SingerWriter()
-    message = RecordMessage(
-        stream="test",
-        record={"id": 1, "name": "test"},
-    )
-    with redirect_stdout(io.StringIO()) as out:
-        writer.write_message(message)
-
-    assert out.getvalue() == (
-        '{"type":"RECORD","stream":"test","record":{"id":1,"name":"test"}}\n'
-    )
-
-
-# Benchmark Tests
+from singer_sdk.contrib.msgspec import MsgSpecReader, MsgSpecWriter
+
+# IO Benchmarks
 
 
 @pytest.fixture
 def bench_record():
     return {
         "stream": "users",
+        "type": "RECORD",
         "record": {
             "Id": 1,
             "created_at": "2021-01-01T00:08:00-07:00",
@@ -151,11 +41,11 @@ def bench_encoded_record(bench_record):
     return json.dumps(bench_record)
 
 
-def test_bench_format_message(benchmark, bench_record_message):
+def test_bench_format_message(benchmark, bench_record_message: RecordMessage):
     """Run benchmark for Sink._validator method validate."""
     number_of_runs = 1000
 
-    writer = SingerWriter()
+    writer = MsgSpecWriter()
 
     def run_format_message():
         for record in itertools.repeat(bench_record_message, number_of_runs):
@@ -164,11 +54,10 @@ def test_bench_format_message(benchmark, bench_record_message):
     benchmark(run_format_message)
 
 
-def test_bench_deserialize_json(benchmark, bench_encoded_record):
+def test_bench_deserialize_json(benchmark, bench_encoded_record: str):
     """Run benchmark for Sink._validator method validate."""
     number_of_runs = 1000
-
-    reader = DummyReader()
+    reader = MsgSpecReader()
 
     def run_deserialize_json():
         for record in itertools.repeat(bench_encoded_record, number_of_runs):
diff --git a/tests/samples/test_tap_countries.py b/tests/samples/test_tap_countries.py
index aa2fb777..e78c21fe 100644
--- a/tests/samples/test_tap_countries.py
+++ b/tests/samples/test_tap_countries.py
@@ -116,7 +116,7 @@ def test_batch_mode(outdir):
         },
     )
 
-    buf = io.StringIO()
+    buf = io.TextIOWrapper(io.BytesIO(), encoding="utf-8")
     with redirect_stdout(buf):
         tap.sync_all()
 
diff --git a/tests/_singerlib/__init__.py b/tests/singerlib/__init__.py
similarity index 100%
rename from tests/_singerlib/__init__.py
rename to tests/singerlib/__init__.py
diff --git a/tests/singerlib/encoding/conftest.py b/tests/singerlib/encoding/conftest.py
new file mode 100644
index 00000000..9d48db4f
--- /dev/null
+++ b/tests/singerlib/encoding/conftest.py
@@ -0,0 +1,1 @@
+from __future__ import annotations
diff --git a/tests/singerlib/encoding/test_msgspec.py b/tests/singerlib/encoding/test_msgspec.py
new file mode 100644
index 00000000..2a544b16
--- /dev/null
+++ b/tests/singerlib/encoding/test_msgspec.py
@@ -0,0 +1,128 @@
+"""Test IO operations for msgspec Singer reader and writer."""
+
+from __future__ import annotations
+
+import decimal
+import io
+from contextlib import nullcontext, redirect_stdout
+from textwrap import dedent
+
+import pytest
+
+from singer_sdk._singerlib import RecordMessage
+from singer_sdk._singerlib.exceptions import InvalidInputLine
+from singer_sdk.contrib.msgspec import (
+    MsgSpecReader,
+    MsgSpecWriter,
+    dec_hook,
+    enc_hook,
+)
+
+CALLBACKS = {
+    "RECORD": lambda _: None,
+    "SCHEMA": lambda _: None,
+    "STATE": lambda _: None,
+    "ACTIVATE_VERSION": lambda _: None,
+    "BATCH": lambda _: None,
+}
+
+
+@pytest.mark.parametrize(
+    "test_type,test_value,expected_value,expected_type",
+    [
+        pytest.param(
+            int,
+            1,
+            "1",
+            str,
+            id="int-to-str",
+        ),
+    ],
+)
+def test_dec_hook(test_type, test_value, expected_value, expected_type):
+    returned = dec_hook(type=test_type, obj=test_value)
+    returned_type = type(returned)
+
+    assert returned == expected_value
+    assert returned_type == expected_type
+
+
+@pytest.mark.parametrize(
+    "test_value,expected_value",
+    [
+        pytest.param(
+            1,
+            "1",
+            id="int-to-str",
+        ),
+    ],
+)
+def test_enc_hook(test_value, expected_value):
+    returned = enc_hook(obj=test_value)
+
+    assert returned == expected_value
+
+
+@pytest.mark.parametrize(
+    "line,expected,exception",
+    [
+        pytest.param(
+            "not-valid-json",
+            None,
+            pytest.raises(InvalidInputLine),
+            id="unparsable",
+        ),
+        pytest.param(
+            '{"type": "RECORD", "stream": "users", "record": {"id": 1, "value": 1.23}}',
+            {
+                "type": "RECORD",
+                "stream": "users",
+                "record": {"id": 1, "value": decimal.Decimal("1.23")},
+            },
+            nullcontext(),
+            id="record",
+        ),
+    ],
+)
+def test_deserialize(line, expected, exception):
+    reader = MsgSpecReader()
+    with exception:
+        assert reader.deserialize_json(line) == expected
+
+
+def test_process_lines():
+    reader = MsgSpecReader()
+    input_lines = io.StringIO(
+        dedent("""\
+        {"type": "SCHEMA", "stream": "users", "schema": {"type": "object", "properties": {"id": {"type": "integer"}, "value": {"type": "number"}}}}
+        {"type": "RECORD", "stream": "users", "record": {"id": 1, "value": 1.23}}
+        {"type": "RECORD", "stream": "users", "record": {"id": 2, "value": 2.34}}
+        {"type": "STATE", "value": {"bookmarks": {"users": {"id": 2}}}}
+        {"type": "SCHEMA", "stream": "batches", "schema": {"type": "object", "properties": {"id": {"type": "integer"}, "value": {"type": "number"}}}}
+        {"type": "BATCH", "stream": "batches", "encoding": {"format": "jsonl", "compression": "gzip"}, "manifest": ["file1.jsonl.gz", "file2.jsonl.gz"]}
+        {"type": "STATE", "value": {"bookmarks": {"users": {"id": 2}, "batches": {"id": 1000000}}}}
+    """)  # noqa: E501
+    )
+    reader.process_lines(input_lines, CALLBACKS)
+
+
+def test_process_unknown_message():
+    reader = MsgSpecReader()
+    input_lines = io.StringIO('{"type": "UNKNOWN"}\n')
+    with pytest.raises(ValueError, match="Unknown message type"):
+        reader.process_lines(input_lines, CALLBACKS)
+
+
+def test_write_message():
+    writer = MsgSpecWriter()
+    message = RecordMessage(
+        stream="test",
+        record={"id": 1, "name": "test"},
+    )
+    with redirect_stdout(io.TextIOWrapper(io.BytesIO())) as out:  # noqa: PLW1514
+        writer.write_message(message)
+
+    out.seek(0)
+    assert out.read() == (
+        '{"type":"RECORD","stream":"test","record":{"id":1,"name":"test"}}\n'
+    )
diff --git a/tests/singerlib/encoding/test_simple.py b/tests/singerlib/encoding/test_simple.py
new file mode 100644
index 00000000..7ce7f5bd
--- /dev/null
+++ b/tests/singerlib/encoding/test_simple.py
@@ -0,0 +1,106 @@
+"""Test IO operations for simple Singer reader and writer."""
+
+from __future__ import annotations
+
+import decimal
+import io
+import json
+from contextlib import nullcontext, redirect_stdout
+from textwrap import dedent
+
+import pytest
+
+from singer_sdk._singerlib import RecordMessage
+from singer_sdk._singerlib.exceptions import InvalidInputLine
+from singer_sdk.singerlib.encoding.simple import (
+    SimpleSingerReader,
+    SimpleSingerWriter,
+)
+
+CALLBACKS = {
+    "RECORD": lambda _: None,
+    "SCHEMA": lambda _: None,
+    "STATE": lambda _: None,
+    "ACTIVATE_VERSION": lambda _: None,
+    "BATCH": lambda _: None,
+}
+
+
+@pytest.mark.parametrize(
+    "line,expected,exception",
+    [
+        pytest.param(
+            "not-valid-json",
+            None,
+            pytest.raises(InvalidInputLine),
+            id="unparsable",
+        ),
+        pytest.param(
+            '{"type": "RECORD", "stream": "users", "record": {"id": 1, "value": 1.23}}',
+            {
+                "type": "RECORD",
+                "stream": "users",
+                "record": {"id": 1, "value": decimal.Decimal("1.23")},
+            },
+            nullcontext(),
+            id="record",
+        ),
+    ],
+)
+def test_deserialize(line, expected, exception):
+    reader = SimpleSingerReader()
+    with exception:
+        assert reader.deserialize_json(line) == expected
+
+
+def test_process_lines():
+    reader = SimpleSingerReader()
+    input_lines = io.StringIO(
+        dedent("""\
+        {"type": "SCHEMA", "stream": "users", "schema": {"type": "object", "properties": {"id": {"type": "integer"}, "value": {"type": "number"}}}}
+        {"type": "RECORD", "stream": "users", "record": {"id": 1, "value": 1.23}}
+        {"type": "RECORD", "stream": "users", "record": {"id": 2, "value": 2.34}}
+        {"type": "STATE", "value": {"bookmarks": {"users": {"id": 2}}}}
+        {"type": "SCHEMA", "stream": "batches", "schema": {"type": "object", "properties": {"id": {"type": "integer"}, "value": {"type": "number"}}}}
+        {"type": "BATCH", "stream": "batches", "encoding": {"format": "jsonl", "compression": "gzip"}, "manifest": ["file1.jsonl.gz", "file2.jsonl.gz"]}
+        {"type": "STATE", "value": {"bookmarks": {"users": {"id": 2}, "batches": {"id": 1000000}}}}
+    """)  # noqa: E501
+    )
+    reader.process_lines(input_lines, CALLBACKS)
+
+
+def test_process_unknown_message():
+    reader = SimpleSingerReader()
+    input_lines = io.StringIO('{"type": "UNKNOWN"}\n')
+    with pytest.raises(ValueError, match="Unknown message type"):
+        reader.process_lines(input_lines, CALLBACKS)
+
+
+def test_process_error():
+    def error(message_dict: dict) -> None:  # noqa: ARG001
+        msg = "Bad record"
+        raise ValueError(msg)
+
+    message = RecordMessage(
+        stream="users",
+        record={"id": 1, "value": 1.23},
+    )
+    input_lines = io.StringIO(json.dumps(message.to_dict()) + "\n")
+    reader = SimpleSingerReader()
+
+    with pytest.raises(ValueError, match="Bad record"):
+        reader.process_lines(input_lines, {**CALLBACKS, "RECORD": error})
+
+
+def test_write_message():
+    writer = SimpleSingerWriter()
+    message = RecordMessage(
+        stream="test",
+        record={"id": 1, "name": "test"},
+    )
+    with redirect_stdout(io.StringIO()) as out:
+        writer.write_message(message)
+
+    assert out.getvalue() == (
+        '{"type":"RECORD","stream":"test","record":{"id":1,"name":"test"}}\n'
+    )