fix: add reducer for consumer

artemmufazalov · artemmufazalov · commit 4ab65e3fb3dd · 2023-02-17T14:36:28.000+03:00
diff --git a/src/services/api.d.ts b/src/services/api.d.ts
@@ -62,6 +62,10 @@ interface Window {
         getTopic: (params: {
             path?: string;
         }) => Promise<import('../types/api/topic').DescribeTopicResult>;
+        getConsumer: (params: {
+            path?: string;
+            consumer?: string;
+        }) => Promise<import('../types/api/consumer').DescribeConsumerResult>;
         [method: string]: Function;
     };
 }
diff --git a/src/services/api.js b/src/services/api.js
@@ -141,12 +141,28 @@ export class YdbEmbeddedAPI extends AxiosWrapper {
             path,
         });
     }
-    getTopic({path}) {
-        return this.get(this.getPath('/viewer/json/describe_topic'), {
-            enums: true,
-            include_stats: true,
-            path,
-        });
+    getTopic({path}, {concurrentId} = {}) {
+        return this.get(
+            this.getPath('/viewer/json/describe_topic'),
+            {
+                enums: true,
+                include_stats: true,
+                path,
+            },
+            {concurrentId: concurrentId || 'getTopic'},
+        );
+    }
+    getConsumer({path, consumer}, {concurrentId} = {}) {
+        return this.get(
+            this.getPath('/viewer/json/describe_consumer'),
+            {
+                enums: true,
+                include_stats: true,
+                path,
+                consumer,
+            },
+            {concurrentId: concurrentId || 'getConsumer'},
+        );
     }
     getPoolInfo(poolName) {
         return this.get(this.getPath('/viewer/json/storage'), {
diff --git a/src/store/reducers/consumer.ts b/src/store/reducers/consumer.ts
@@ -0,0 +1,160 @@
+/* eslint-disable camelcase */
+import type {Reducer} from 'redux';
+import {createSelector, Selector} from 'reselect';
+
+import type {
+    IConsumerAction,
+    IConsumerRootStateSlice,
+    IConsumerState,
+    IPreparedPartitionData,
+} from '../../types/store/consumer';
+
+import '../../services/api';
+
+import {convertBytesObjectToSpeed} from '../../utils/bytesParsers';
+import {parseLag, parseTimestampToIdleTime} from '../../utils/timeParsers';
+import {isNumeric} from '../../utils/utils';
+
+import {createRequestActionTypes, createApiRequest} from '../utils';
+
+export const FETCH_CONSUMER = createRequestActionTypes('consumer', 'FETCH_CONSUMER');
+
+const SET_DATA_WAS_NOT_LOADED = 'consumer/SET_DATA_WAS_NOT_LOADED';
+
+const initialState = {
+    loading: false,
+    wasLoaded: false,
+    data: {},
+};
+
+const consumer: Reducer<IConsumerState, IConsumerAction> = (state = initialState, action) => {
+    switch (action.type) {
+        case FETCH_CONSUMER.REQUEST: {
+            return {
+                ...state,
+                loading: true,
+            };
+        }
+        case FETCH_CONSUMER.SUCCESS: {
+            // On older version it can return HTML page of Internal Viewer with an error
+            if (typeof action.data !== 'object') {
+                return {...state, loading: false, error: {}};
+            }
+
+            return {
+                ...state,
+                data: action.data,
+                loading: false,
+                wasLoaded: true,
+                error: undefined,
+            };
+        }
+        case FETCH_CONSUMER.FAILURE: {
+            if (action.error?.isCancelled) {
+                return state;
+            }
+
+            return {
+                ...state,
+                error: action.error,
+                loading: false,
+            };
+        }
+        case SET_DATA_WAS_NOT_LOADED: {
+            return {
+                ...state,
+                wasLoaded: false,
+            };
+        }
+        default:
+            return state;
+    }
+};
+
+export const setDataWasNotLoaded = () => {
+    return {
+        type: SET_DATA_WAS_NOT_LOADED,
+    } as const;
+};
+
+export function getConsumer(path?: string, consumerName?: string) {
+    return createApiRequest({
+        request: window.api.getConsumer({path, consumer: consumerName}),
+        actions: FETCH_CONSUMER,
+    });
+}
+
+export const selectPartitions = (state: IConsumerRootStateSlice) => state.consumer.data?.partitions;
+
+export const selectPreparedPartitionsData: Selector<
+    IConsumerRootStateSlice,
+    IPreparedPartitionData[] | undefined
+> = createSelector([selectPartitions], (partitions) => {
+    return partitions?.map((partition) => {
+        // describe_consumer endpoint doesn't return zero values, so some values will be initialized with 0
+        const {partition_id = '0', partition_stats, partition_consumer_stats} = partition;
+
+        const {
+            partition_offsets,
+            store_size_bytes = '0',
+            last_write_time: partition_last_write_time,
+            max_write_time_lag: partition_write_lag,
+            bytes_written,
+            partition_node_id = 0,
+        } = partition_stats || {};
+
+        const {start: start_offset = '0', end: end_offset = '0'} = partition_offsets || {};
+
+        const {
+            last_read_offset = '0',
+            committed_offset = '0',
+            read_session_id,
+            last_read_time: consumer_last_read_time,
+            max_read_time_lag: consumer_read_lag,
+            max_write_time_lag: consumer_write_lag,
+            bytes_read,
+            reader_name,
+            connection_node_id = 0,
+        } = partition_consumer_stats || {};
+
+        const uncommitedMessages =
+            isNumeric(end_offset) && isNumeric(committed_offset)
+                ? Number(end_offset) - Number(committed_offset)
+                : 0;
+
+        const unreadMessages =
+            isNumeric(end_offset) && isNumeric(last_read_offset)
+                ? Number(end_offset) - Number(last_read_offset)
+                : 0;
+
+        return {
+            partitionId: partition_id,
+            storeSize: store_size_bytes,
+
+            writeSpeed: convertBytesObjectToSpeed(bytes_written),
+            readSpeed: convertBytesObjectToSpeed(bytes_read),
+
+            partitionWriteLag: parseLag(partition_write_lag),
+            partitionWriteIdleTime: parseTimestampToIdleTime(partition_last_write_time),
+
+            consumerWriteLag: parseLag(consumer_write_lag),
+            consumerReadLag: parseLag(consumer_read_lag),
+            consumerReadIdleTime: parseTimestampToIdleTime(consumer_last_read_time),
+
+            uncommitedMessages,
+            unreadMessages,
+
+            startOffset: start_offset,
+            endOffset: end_offset,
+            commitedOffset: committed_offset,
+
+            readSessionId: read_session_id,
+            readerName: reader_name,
+
+            partitionNodeId: partition_node_id,
+            connectionNodeId: connection_node_id,
+        };
+    });
+});
+
+export default consumer;
diff --git a/src/store/reducers/index.ts b/src/store/reducers/index.ts
@@ -18,6 +18,7 @@ import pool from './pool';
 import tenants from './tenants';
 import tablet from './tablet';
 import topic from './topic';
+import consumer from './consumer';
 import executeQuery from './executeQuery';
 import explainQuery from './explainQuery';
 import tabletsFilters from './tabletsFilters';
@@ -57,6 +58,7 @@ export const rootReducer = {
     tenants,
     tablet,
     topic,
+    consumer,
     executeQuery,
     explainQuery,
     tabletsFilters,
diff --git a/src/store/reducers/topic.ts b/src/store/reducers/topic.ts
@@ -7,10 +7,12 @@ import {createRequestActionTypes, createApiRequest} from '../utils';
 
 export const FETCH_TOPIC = createRequestActionTypes('topic', 'FETCH_TOPIC');
 
+const SET_DATA_WAS_NOT_LOADED = 'topic/SET_DATA_WAS_NOT_LOADED';
+
 const initialState = {
     loading: true,
     wasLoaded: false,
-    data: {},
+    data: undefined,
 };
 
 const topic: Reducer<ITopicState, ITopicAction> = (state = initialState, action) => {
@@ -22,6 +24,11 @@ const topic: Reducer<ITopicState, ITopicAction> = (state = initialState, action)
             };
         }
         case FETCH_TOPIC.SUCCESS: {
+            // On older version it can return HTML page of Internal Viewer with an error
+            if (typeof action.data !== 'object') {
+                return {...state, loading: false, error: {}};
+            }
+
             return {
                 ...state,
                 data: action.data,
@@ -31,17 +38,33 @@ const topic: Reducer<ITopicState, ITopicAction> = (state = initialState, action)
             };
         }
         case FETCH_TOPIC.FAILURE: {
+            if (action.error?.isCancelled) {
+                return state;
+            }
+
             return {
                 ...state,
                 error: action.error,
                 loading: false,
             };
         }
+        case SET_DATA_WAS_NOT_LOADED: {
+            return {
+                ...state,
+                wasLoaded: false,
+            };
+        }
         default:
             return state;
     }
 };
 
+export const setDataWasNotLoaded = () => {
+    return {
+        type: SET_DATA_WAS_NOT_LOADED,
+    } as const;
+};
+
 export function getTopic(path?: string) {
     return createApiRequest({
         request: window.api.getTopic({path}),
diff --git a/src/types/store/consumer.ts b/src/types/store/consumer.ts
@@ -0,0 +1,55 @@
+import type {IProcessSpeedStats} from '../../utils/bytesParsers';
+import type {ApiRequestAction} from '../../store/utils';
+
+import {FETCH_CONSUMER, setDataWasNotLoaded} from '../../store/reducers/consumer';
+
+import type {DescribeConsumerResult} from '../api/consumer';
+import type {IResponseError} from '../api/error';
+
+// All fields should be present though they could be undefined
+export interface IPreparedPartitionData {
+    partitionId: string;
+    storeSize: string;
+
+    writeSpeed: IProcessSpeedStats;
+    readSpeed: IProcessSpeedStats;
+
+    partitionWriteLag: number;
+    partitionWriteIdleTime: number;
+
+    consumerWriteLag: number;
+    consumerReadLag: number;
+    consumerReadIdleTime: number;
+
+    uncommitedMessages: number;
+    unreadMessages: number;
+
+    startOffset: string;
+    endOffset: string;
+    commitedOffset: string;
+
+    readSessionId: string | undefined;
+    readerName: string | undefined;
+
+    partitionNodeId: number;
+    connectionNodeId: number;
+}
+
+export interface IConsumerState {
+    loading: boolean;
+    wasLoaded: boolean;
+    data?: DescribeConsumerResult;
+    error?: IResponseError;
+}
+
+type IConsumerApiRequestAction = ApiRequestAction<
+    typeof FETCH_CONSUMER,
+    DescribeConsumerResult,
+    IResponseError
+>;
+
+export type IConsumerAction = IConsumerApiRequestAction | ReturnType<typeof setDataWasNotLoaded>;
+
+export interface IConsumerRootStateSlice {
+    consumer: IConsumerState;
+}
diff --git a/src/types/store/topic.ts b/src/types/store/topic.ts
@@ -1,4 +1,4 @@
-import {FETCH_TOPIC} from '../../store/reducers/topic';
+import {FETCH_TOPIC, setDataWasNotLoaded} from '../../store/reducers/topic';
 import type {ApiRequestAction} from '../../store/utils';
 import type {IResponseError} from '../api/error';
 import type {DescribeTopicResult} from '../api/topic';
@@ -10,11 +10,9 @@ export interface ITopicState {
     error?: IResponseError;
 }
 
-export type ITopicAction = ApiRequestAction<
-    typeof FETCH_TOPIC,
-    DescribeTopicResult,
-    IResponseError
->;
+export type ITopicAction =
+    | ApiRequestAction<typeof FETCH_TOPIC, DescribeTopicResult, IResponseError>
+    | ReturnType<typeof setDataWasNotLoaded>;
 
 export interface ITopicRootStateSlice {
     topic: ITopicState;
