diff --git a/packages/rpc-subscriptions/src/rpc-subscriptions-json.ts b/packages/rpc-subscriptions/src/rpc-subscriptions-json.ts
index 2a3d1d7032b4..9315b49a34d9 100644
--- a/packages/rpc-subscriptions/src/rpc-subscriptions-json.ts
+++ b/packages/rpc-subscriptions/src/rpc-subscriptions-json.ts
@@ -1,26 +1,16 @@
-import { RpcSubscriptionsChannel } from '@solana/rpc-subscriptions-spec';
+import { pipe } from '@solana/functional';
+import {
+    RpcSubscriptionsChannel,
+    transformChannelInboundMessages,
+    transformChannelOutboundMessages,
+} from '@solana/rpc-subscriptions-spec';
 
 export function getRpcSubscriptionsChannelWithJSONSerialization(
     channel: RpcSubscriptionsChannel<string, string>,
 ): RpcSubscriptionsChannel<unknown, unknown> {
-    return Object.freeze({
-        ...channel,
-        on(type, listener, options) {
-            if (type !== 'message') {
-                return channel.on(type, listener, options);
-            }
-            return channel.on(
-                'message',
-                function deserializingListener(message: string) {
-                    const deserializedMessage = JSON.parse(message);
-                    listener(deserializedMessage);
-                },
-                options,
-            );
-        },
-        send(message) {
-            const serializedMessage = JSON.stringify(message);
-            return channel.send(serializedMessage);
-        },
-    } as RpcSubscriptionsChannel<unknown, unknown>);
+    return pipe(
+        channel,
+        c => transformChannelInboundMessages(c, JSON.parse),
+        c => transformChannelOutboundMessages(c, JSON.stringify),
+    );
 }