javascript - 如何在收到 websocket 消息时更新从 RTK 查询返回的 Hook 数据

Question

有人可以解释一下如何将 websocket 消息中的数据接收连接到来自 RTK 查询 api 端点的钩子(Hook)的数据对象吗？

我们不需要存储收到的消息，我们只想将其传递到 useGetWebsocketResponseQuery Hook 的数据参数中，以便我们可以在 UI 中触发通知。

reducerPath: 'someApi',
    baseQuery: baseQueryWithReauth,
    endpoints: (builder) => ({
        getWebsocketResponse: builder.query<WebsocketResult, void>({
            queryFn: () => ({data: {}),
            async onCacheEntryAdded(arg, { updateCachedData, cacheDataLoaded, cacheEntryRemoved }) {        
                try {                  
                    // wait for the initial query to resolve before proceeding
                    await cacheDataLoaded;

                    const socket = io('http://url', {});                
                    socket.on('connect', () => {
                        console.log('socket connected on rtk query');
                    });

                    socket.on('message', (message) => {
                        console.log(`received message: ${message}`);
                        // THIS IS WHERE THE DATA NEEDS TO BE WIRED UP TO THE HOOK BUT HOW?
                    });

                    await cacheEntryRemoved;        
                } catch {
                    // no-op in case `cacheEntryRemoved` resolves before `cacheDataLoaded`,
                    // in which case `cacheDataLoaded` will throw
                }                 
            }
        }),
// Export hooks for usage in functional components, which are
// auto-generated based on the defined endpoints
export const {
    useGetWebsocketResponseQuery
} = someApi;

感觉这在 socket.on('message', {}) 处理程序中应该是可能的，但是怎么做呢？ updateCachedData 方法似乎应该是可行的方法，但我不确定如何实现它。

感谢所有帮助:-)

非常感谢，

山姆

---

更新来自@phry 的解决方案

问题是 queryFn 中定义的数据需要匹配从 cacheDataLoaded 附加的数据的形状，即它应该是这样的:-

queryFn: () => ({data: { messages: [] }),

和

socket.on('connect', () => {                        
   updateCachedData((currentCacheData) => {
      currentCacheData.messages.push(message);
   });
});

Answer 1

在阅读 redux 工具包文档后，我发现到目前为止你应该使用 createEntityAdapter为缓存数据创建数据库架构，并在从套接字收到消息后使用 CRUD 更新该数据库(缓存)函数，例如，首先触发 API 查询并获取数据，您可以使用 createEntityAdapter 的 addOne 或 addMany，然后当从套接字接收到数据时您可以使用 setOne 来为 setMany 更新 updateCachedData 中的缓存数据。通过这样做，use[yourWhatEver]Query 中的实际数据首先在查询获取 API 时更新，然后每次从套接字接收数据时更新。

import { createApi } from '@reduxjs/toolkit/query/react';
import axiosBaseQuery from 'api/axiosBaseQuery';
import { createEntityAdapter } from '@reduxjs/toolkit';

const instrumentsAdapter = createEntityAdapter({
    selectId: (item) => item?.state?.symbol
});

export const marketApi = createApi({
    reducerPath: 'api/market',
    baseQuery: axiosBaseQuery(),
    endpoints: (builder) => ({
        getInstrumentByRefId: builder.query({
            query: (refId) => ({
                url: `/market/instruments/${refId}/summary`,
                method: 'get'
            }),
            transformResponse: (res) => {
                return instrumentsAdapter.addMany(instrumentsAdapter.getInitialState(), [res]);
            },
            async onCacheEntryAdded(arg, { updateCachedData, cacheDataLoaded, cacheEntryRemoved }) {
                // arg === refId
                const payload = `instruments.${arg}.summary`;

                // create a websocket connection when the cache subscription starts
                const ws = new WebSocket('wss://[domain.com]/api/notification/ws');
                let waitTimer = null;
                const waitForConnection = (callback, interval) => {
                    clearTimeout(waitTimer);

                    if (ws && ws.readyState === 1) {
                        callback();
                        return;
                    }

                    // optional: implement backoff for interval here
                    waitTimer = setTimeout(() => {
                        waitForConnection(callback, interval);
                    }, interval);
                };
                try {
                    // wait for the initial query to resolve before proceeding
                    await cacheDataLoaded;

                    // when data is received from the socket connection to the server,
                    // if it is a message and for the appropriate channel,
                    // update our query result with the received message
                    const listener = (event) => {
                        const data = JSON.parse(event.data);
                        // eslint-disable-next-line no-console
                        // console.log('data', data);
                        // if (!isMessage(data) || data.channel !== arg) return;
                        updateCachedData((draft) => {
                            // eslint-disable-next-line no-unused-vars, no-param-reassign
                            if (data.value) {
                                instrumentsAdapter.setMany(draft, [data.value]);
                            }
                        });
                    };

                    waitForConnection(() => {
                        ws.send(
                            JSON.stringify({
                                id: '1',
                                type: 'SUBSCRIBE',
                                path: payload
                            })
                        );
                    }, 100);

                    ws.onmessage = listener;
                    // ws.addEventListener('message', listener);
                } catch (err) {
                    console.log('err', err);
                    // no-op in case `cacheEntryRemoved` resolves before `cacheDataLoaded`,
                    // in which case `cacheDataLoaded` will throw
                }
                // cacheEntryRemoved will resolve when the cache subscription is no longer active
                await cacheEntryRemoved;
                // perform cleanup steps once the `cacheEntryRemoved` promise resolves
                ws.close();
            }
        }),
        getCandles: builder.query({
            query: ({ refId, bucket, end, limit = 1 }) => ({
                url: `/market/instruments/${refId}/candles?bucket=${bucket}&end=${end}&limit=${limit}`,
                method: 'get'
            })
        })
    })
});

export const {
    useGetMarketMapQuery,
    useGetInstrumentByRefIdQuery,
    useGetInstrumentsQuery,
    useGetCandlesQuery
} = marketApi;

在 React 组件中你可以更新值

function MyReactComponent() {
    // data will gets updated every time socket data received.
    const { data } = useGetInstrumentByRefIdQuery('IRO1SIPA0001');

    return JSON.stringify(data);
}

希望对您有所帮助。