Subscription error: TRPCClientError: Subscriptions should use wsLink

this is my trpc client:

export const trpc = createTRPCNext<AppRouter>({
  /**
   * @link https://trpc.io/docs/v11/ssr
   */
  // ssr: true,
  // ssrPrepass,
  ssr: false,
  config({ ctx }) {
    /**
     * If you want to use SSR, you need to use the server's full URL
     * @link https://trpc.io/docs/v11/ssr
     */

    return {
      /**
       * @link https://trpc.io/docs/v11/client/links
       */
      links: [
        // adds pretty logs to your console in development and logs errors in production
        loggerLink({
          enabled: opts =>
            (process.env.NODE_ENV === 'development' && typeof window !== 'undefined') ||
            (opts.direction === 'down' && opts.result instanceof Error)
        }),
        splitLink({
          condition: op => {
            return op.type === 'subscription'
          },
          true: wsLink({
            client: createWSClient({
              url: 'ws://localhost:3001'
            })
          }),
          false: httpBatchLink({
            transformer: superjson,
            url: `http://localhost:3000/api/trpc`
          })
        })
      ],
      /**
       * @link https://tanstack.com/query/v5/docs/reference/QueryClient
       */
      queryClientConfig: { defaultOptions: { queries: { staleTime: 60 } } }
    }
  },
  /**
   * @link https://trpc.io/docs/v11/data-transformers
   */
  transformer: superjson
})

export const trpc = createTRPCNext<AppRouter>({
  /**
   * @link https://trpc.io/docs/v11/ssr
   */
  // ssr: true,
  // ssrPrepass,
  ssr: false,
  config({ ctx }) {
    /**
     * If you want to use SSR, you need to use the server's full URL
     * @link https://trpc.io/docs/v11/ssr
     */

    return {
      /**
       * @link https://trpc.io/docs/v11/client/links
       */
      links: [
        // adds pretty logs to your console in development and logs errors in production
        loggerLink({
          enabled: opts =>
            (process.env.NODE_ENV === 'development' && typeof window !== 'undefined') ||
            (opts.direction === 'down' && opts.result instanceof Error)
        }),
        splitLink({
          condition: op => {
            return op.type === 'subscription'
          },
          true: wsLink({
            client: createWSClient({
              url: 'ws://localhost:3001'
            })
          }),
          false: httpBatchLink({
            transformer: superjson,
            url: `http://localhost:3000/api/trpc`
          })
        })
      ],
      /**
       * @link https://tanstack.com/query/v5/docs/reference/QueryClient
       */
      queryClientConfig: { defaultOptions: { queries: { staleTime: 60 } } }
    }
  },
  /**
   * @link https://trpc.io/docs/v11/data-transformers
   */
  transformer: superjson
})

I don't know what is the issue exactly.

14 Replies

mithrandir•3mo ago

anyone help please? im stuck and I dont know what shopuld I do here, heres a stackoverflow explaination of my issue https://stackoverflow.com/questions/78653455/error-subscriptions-should-use-wslink-using-trpc

Stack Overflow

Error "Subscriptions should use wsLink" using tRPC

While trying to implement subscriptions with tRPC, I encountered this issue, that it should use wsLink, even though, it is implemented in the trpc client. I get this error when trying to use random...

BeBoRE•3mo ago

This example and the stack overflow are different, the example here should work

mithrandir•3mo ago

it doesn't actually I am not sure where I am wrong here is a detailed vers:

"dev": "run-p dev:*",
    "dev:next": "next dev",
    "dev:wss": "cross-env PORT=3001 nodemon --watch src --ext .ts,.tsx,js,jsx --signal SIGTERM --exec \"ts-node --project tsconfig.server.json src/server/wsServer.ts\"",

"dev": "run-p dev:*",
    "dev:next": "next dev",
    "dev:wss": "cross-env PORT=3001 nodemon --watch src --ext .ts,.tsx,js,jsx --signal SIGTERM --exec \"ts-node --project tsconfig.server.json src/server/wsServer.ts\"",

export const trpc = createTRPCNext<AppRouter>({
  /**
   * @link https://trpc.io/docs/v11/ssr
   */
  // ssr: true,
  // ssrPrepass,
  ssr: false,
  config({ ctx }) {
    /**
     * If you want to use SSR, you need to use the server's full URL
     * @link https://trpc.io/docs/v11/ssr
     */

    return {
      /**
       * @link https://trpc.io/docs/v11/client/links
       */
      links: [
        // adds pretty logs to your console in development and logs errors in production
        loggerLink({
          enabled: opts =>
            (process.env.NODE_ENV === 'development' && typeof window !== 'undefined') ||
            (opts.direction === 'down' && opts.result instanceof Error)
        }),
        splitLink({
          condition: op => {
            return op.type === 'subscription'
          },
          true: wsLink({
            client: createWSClient({
              url: 'ws://localhost:3001'
            })
          }),
          false: httpBatchLink({
            transformer: superjson,
            url: `http://localhost:3000/api/trpc`
          })
        })
      ],
      /**
       * @link https://tanstack.com/query/v5/docs/reference/QueryClient
       */
      queryClientConfig: { defaultOptions: { queries: { staleTime: 60 } } }
    }
  },
  /**
   * @link https://trpc.io/docs/v11/data-transformers
   */
  transformer: superjson
})

export const trpc = createTRPCNext<AppRouter>({
  /**
   * @link https://trpc.io/docs/v11/ssr
   */
  // ssr: true,
  // ssrPrepass,
  ssr: false,
  config({ ctx }) {
    /**
     * If you want to use SSR, you need to use the server's full URL
     * @link https://trpc.io/docs/v11/ssr
     */

    return {
      /**
       * @link https://trpc.io/docs/v11/client/links
       */
      links: [
        // adds pretty logs to your console in development and logs errors in production
        loggerLink({
          enabled: opts =>
            (process.env.NODE_ENV === 'development' && typeof window !== 'undefined') ||
            (opts.direction === 'down' && opts.result instanceof Error)
        }),
        splitLink({
          condition: op => {
            return op.type === 'subscription'
          },
          true: wsLink({
            client: createWSClient({
              url: 'ws://localhost:3001'
            })
          }),
          false: httpBatchLink({
            transformer: superjson,
            url: `http://localhost:3000/api/trpc`
          })
        })
      ],
      /**
       * @link https://tanstack.com/query/v5/docs/reference/QueryClient
       */
      queryClientConfig: { defaultOptions: { queries: { staleTime: 60 } } }
    }
  },
  /**
   * @link https://trpc.io/docs/v11/data-transformers
   */
  transformer: superjson
})

import ws from 'ws'
import { applyWSSHandler } from '@trpc/server/adapters/ws'
import { appRouter } from './routers/_app'
import { createContext, createTRPCContext } from './context'

const wss = new ws.Server({
  port: 3001
})

const handler = applyWSSHandler({ wss, router: appRouter, createContext: createTRPCContext })

wss.on('connection', () => {
  console.log(`Got a connection ${wss.clients.size}`)
  wss.once('close', () => {
    console.log(`Closed connection ${wss.clients.size}`)
  })
})

console.log(`wss server start at ws://localhost:3001`)

process.on('SIGTERM', () => {
  console.log('Got SIGTERM')
  handler.broadcastReconnectNotification()
  wss.close()
})

import ws from 'ws'
import { applyWSSHandler } from '@trpc/server/adapters/ws'
import { appRouter } from './routers/_app'
import { createContext, createTRPCContext } from './context'

const wss = new ws.Server({
  port: 3001
})

const handler = applyWSSHandler({ wss, router: appRouter, createContext: createTRPCContext })

wss.on('connection', () => {
  console.log(`Got a connection ${wss.clients.size}`)
  wss.once('close', () => {
    console.log(`Closed connection ${wss.clients.size}`)
  })
})

console.log(`wss server start at ws://localhost:3001`)

process.on('SIGTERM', () => {
  console.log('Got SIGTERM')
  handler.broadcastReconnectNotification()
  wss.close()
})

onAdd: publicProcedure.subscription(() => {
    // return an `observable` with a callback which is triggered immediately
    return observable<Review>(emit => {
      const onAdd = (data: Review) => {
        // emit data to client
        emit.next(data)
      }
      // trigger `onAdd()` when `add` is triggered in our event emitter
      ee.on('add', onAdd)
      // unsubscribe function when client disconnects or stops subscribing
      return () => {
        ee.off('add', onAdd)
      }
    })
  }),
  createReview: authedProcedure
    .input(v => {
      const schema = z.object({
       ...
      })
      const result = schema.safeParse(v)
      if (!result.success) {
        throw result.error
      }
      return result.data
    })
    .mutation(async ({ input, ctx }) => {
      const session = await getServerAuthSession()

      // Create the review record
      const review = await db.review.create({
        data: {
          ....
        },
     
      })
      ee.emit('add', review)
      return review
    }),

onAdd: publicProcedure.subscription(() => {
    // return an `observable` with a callback which is triggered immediately
    return observable<Review>(emit => {
      const onAdd = (data: Review) => {
        // emit data to client
        emit.next(data)
      }
      // trigger `onAdd()` when `add` is triggered in our event emitter
      ee.on('add', onAdd)
      // unsubscribe function when client disconnects or stops subscribing
      return () => {
        ee.off('add', onAdd)
      }
    })
  }),
  createReview: authedProcedure
    .input(v => {
      const schema = z.object({
       ...
      })
      const result = schema.safeParse(v)
      if (!result.success) {
        throw result.error
      }
      return result.data
    })
    .mutation(async ({ input, ctx }) => {
      const session = await getServerAuthSession()

      // Create the review record
      const review = await db.review.create({
        data: {
          ....
        },
     
      })
      ee.emit('add', review)
      return review
    }),

mithrandir•3mo ago

when i console log i get this err :

BeBoRE•3mo ago

Yeah idk, looking at the example the code should work in the browser. Is there a ws connection being made?

mithrandir•3mo ago

sorry i dont understand ur qst @Alex / KATT 🐱 any ideas pleasse? been stuck on this for quite sometime

Alex / KATT 🐱•3mo ago

use https://trpc.io/docs/client/links/httpSubscriptionLink instead, it's easier

HTTP Subscription Link | tRPC

httpSubscriptionLink is a terminating link that's uses Server-sent Events (SSE) for subscriptions.

mithrandir•3mo ago

Module '"@trpc/client"' has no exported member 'unstable_httpSubscriptionLink'.ts(2305) @Alex / KATT 🐱 what to do here?

BeBoRE•3mo ago

Use the Next branch if you want to use that. I don’t think it’ll solve your issue, because it will still be attempting to use your htthBatchLink for subscriptions somewhere

mithrandir•3mo ago

if you mean this, I've already added it, but still doesn't get imported

Alex / KATT 🐱•3mo ago

npm install @trpc/server@next @trpc/client@next @trpc/react-query@next @trpc/next@next @tanstack/react-query@latest @tanstack/react-query-devtools@latest

mithrandir•3mo ago

I have a question, @Alex / KATT 🐱 the subscription works when only I replace wsLink(...) with unstable_httpSubscriptionLink why is that? in the client provider

Alex / KATT 🐱•3mo ago

you don't use websockets so something is prob wrong with your server setup just kill all your websocket server stuff and use the httpSubscriptionLink

Alex / KATT 🐱•3mo ago

https://trpc.io/docs/client/links/httpSubscriptionLink

HTTP Subscription Link | tRPC

httpSubscriptionLink is a terminating link that's uses Server-sent Events (SSE) for subscriptions.