mithrandir
mithrandir•5mo ago

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
mithrandirOP•5mo 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
BeBoRE•5mo ago
This example and the stack overflow are different, the example here should work
mithrandir
mithrandirOP•5mo 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
mithrandirOP•5mo ago
when i console log i get this err :
No description
BeBoRE
BeBoRE•5mo ago
Yeah idk, looking at the example the code should work in the browser. Is there a ws connection being made?
mithrandir
mithrandirOP•5mo ago
sorry i dont understand ur qst @Alex / KATT 🐱 any ideas pleasse? been stuck on this for quite sometime
Alex / KATT 🐱
Alex / KATT 🐱•5mo 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
mithrandirOP•5mo ago
Module '"@trpc/client"' has no exported member 'unstable_httpSubscriptionLink'.ts(2305) @Alex / KATT 🐱 what to do here?
BeBoRE
BeBoRE•5mo 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
mithrandirOP•5mo ago
if you mean this, I've already added it, but still doesn't get imported
No description
Alex / KATT 🐱
Alex / KATT 🐱•5mo 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
mithrandirOP•5mo 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 🐱
Alex / KATT 🐱•5mo 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 🐱
Alex / KATT 🐱•5mo 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.