Issues/1740 live tailing order (#1824)

* fixing Message addition order during Live tailing

* fixing Message addition order during Live tailing adding tests suites

* minor changes in the function name for prepending addTopics during live mode

* delete minor code repetition

Co-authored-by: Roman Zabaluev <rzabaluev@provectus.com>

kafka-ui-react-app/src/components/Topics/Topic/Details/Messages/Filters/Filters.tsx

@@ -1,6 +1,7 @@
 import 'react-datepicker/dist/react-datepicker.css';
 
 import {
+  MessageFilterType,
   Partition,
   SeekDirection,
   SeekType,
@@ -8,7 +9,6 @@ import {
   TopicMessageConsuming,
   TopicMessageEvent,
   TopicMessageEventTypeEnum,
-  MessageFilterType,
 } from 'generated-sources';
 import React, { useContext } from 'react';
 import { omitBy } from 'lodash';
@@ -17,7 +17,7 @@ import DatePicker from 'react-datepicker';
 import MultiSelect from 'components/common/MultiSelect/MultiSelect.styled';
 import { Option } from 'react-multi-select-component/dist/lib/interfaces';
 import BytesFormatted from 'components/common/BytesFormatted/BytesFormatted';
-import { TopicName, ClusterName } from 'redux/interfaces';
+import { ClusterName, TopicName } from 'redux/interfaces';
 import { BASE_PARAMS } from 'lib/constants';
 import Input from 'components/common/Input/Input';
 import Select from 'components/common/Select/Select';
@@ -45,7 +45,7 @@ export interface FiltersProps {
   partitions: Partition[];
   meta: TopicMessageConsuming;
   isFetching: boolean;
-  addMessage(message: TopicMessage): void;
+  addMessage(content: { message: TopicMessage; prepend: boolean }): void;
   resetMessages(): void;
   updatePhase(phase: string): void;
   updateMeta(meta: TopicMessageConsuming): void;
@@ -304,7 +304,12 @@ const Filters: React.FC<FiltersProps> = ({
 
         switch (type) {
           case TopicMessageEventTypeEnum.MESSAGE:
-            if (message) addMessage(message);
+            if (message) {
+              addMessage({
+                message,
+                prepend: isLive,
+              });
+            }
             break;
           case TopicMessageEventTypeEnum.PHASE:
             if (phase?.name) updatePhase(phase.name);

kafka-ui-react-app/src/redux/actions/__test__/actions.spec.ts

@@ -68,7 +68,7 @@ describe('Actions', () => {
   });
 
   describe('setTopicsSearchAction', () => {
-    it('creartes SET_TOPICS_SEARCH', () => {
+    it('creates SET_TOPICS_SEARCH', () => {
       expect(actions.setTopicsSearchAction('test')).toEqual({
         type: 'SET_TOPICS_SEARCH',
         payload: 'test',
@@ -77,7 +77,7 @@ describe('Actions', () => {
   });
 
   describe('setTopicsOrderByAction', () => {
-    it('creartes SET_TOPICS_ORDER_BY', () => {
+    it('creates SET_TOPICS_ORDER_BY', () => {
       expect(actions.setTopicsOrderByAction(TopicColumnsToSort.NAME)).toEqual({
         type: 'SET_TOPICS_ORDER_BY',
         payload: TopicColumnsToSort.NAME,
@@ -87,10 +87,12 @@ describe('Actions', () => {
 
   describe('topic messages', () => {
     it('creates ADD_TOPIC_MESSAGE', () => {
-      expect(actions.addTopicMessage(topicMessagePayload)).toEqual({
+      expect(actions.addTopicMessage({ message: topicMessagePayload })).toEqual(
+        {
           type: 'ADD_TOPIC_MESSAGE',
-        payload: topicMessagePayload,
-      });
+          payload: { message: topicMessagePayload },
+        }
+      );
     });
 
     it('creates RESET_TOPIC_MESSAGES', () => {

kafka-ui-react-app/src/redux/actions/actions.ts

@@ -161,8 +161,10 @@ export const fetchTopicConsumerGroupsAction = createAsyncAction(
   'GET_TOPIC_CONSUMER_GROUPS__FAILURE'
 )<undefined, TopicsState, undefined>();
 
-export const addTopicMessage =
-  createAction('ADD_TOPIC_MESSAGE')<TopicMessage>();
+export const addTopicMessage = createAction('ADD_TOPIC_MESSAGE')<{
+  message: TopicMessage;
+  prepend?: boolean;
+}>();
 
 export const resetTopicMessages = createAction('RESET_TOPIC_MESSAGES')();
 

kafka-ui-react-app/src/redux/reducers/topicMessages/__test__/fixtures.ts

@@ -15,6 +15,12 @@ export const topicMessagePayload: TopicMessage = {
     '{"host":"schemaregistry1","port":8085,"master_eligibility":true,"scheme":"http","version":1}',
 };
 
+export const topicMessagePayloadV2: TopicMessage = {
+  ...topicMessagePayload,
+  partition: 28,
+  offset: 88,
+};
+
 export const topicMessagesMetaPayload: TopicMessageConsuming = {
   bytesConsumed: 1830,
   elapsedMs: 440,

kafka-ui-react-app/src/redux/reducers/topicMessages/__test__/reducer.spec.ts

@@ -6,16 +6,59 @@ import {
 } from 'redux/actions';
 import reducer from 'redux/reducers/topicMessages/reducer';
 
-import { topicMessagePayload, topicMessagesMetaPayload } from './fixtures';
+import {
+  topicMessagePayload,
+  topicMessagePayloadV2,
+  topicMessagesMetaPayload,
+} from './fixtures';
 
 describe('TopicMessages reducer', () => {
   it('Adds new message', () => {
-    const state = reducer(undefined, addTopicMessage(topicMessagePayload));
+    const state = reducer(
+      undefined,
+      addTopicMessage({ message: topicMessagePayload })
+    );
     expect(state.messages.length).toEqual(1);
     expect(state).toMatchSnapshot();
   });
+
+  it('Adds new message with live tailing one', () => {
+    const state = reducer(
+      undefined,
+      addTopicMessage({ message: topicMessagePayload })
+    );
+    const modifiedState = reducer(
+      state,
+      addTopicMessage({ message: topicMessagePayloadV2, prepend: true })
+    );
+    expect(modifiedState.messages.length).toEqual(2);
+    expect(modifiedState.messages).toEqual([
+      topicMessagePayloadV2,
+      topicMessagePayload,
+    ]);
+  });
+
+  it('Adds new message with live tailing off', () => {
+    const state = reducer(
+      undefined,
+      addTopicMessage({ message: topicMessagePayload })
+    );
+    const modifiedState = reducer(
+      state,
+      addTopicMessage({ message: topicMessagePayloadV2 })
+    );
+    expect(modifiedState.messages.length).toEqual(2);
+    expect(modifiedState.messages).toEqual([
+      topicMessagePayload,
+      topicMessagePayloadV2,
+    ]);
+  });
+
   it('Clears messages', () => {
-    const state = reducer(undefined, addTopicMessage(topicMessagePayload));
+    const state = reducer(
+      undefined,
+      addTopicMessage({ message: topicMessagePayload })
+    );
     expect(state.messages.length).toEqual(1);
 
     const newState = reducer(state, resetTopicMessages());

kafka-ui-react-app/src/redux/reducers/topicMessages/__test__/selectors.spec.ts

@@ -28,7 +28,7 @@ describe('TopicMessages selectors', () => {
 
   describe('state', () => {
     beforeAll(() => {
-      store.dispatch(addTopicMessage(topicMessagePayload));
+      store.dispatch(addTopicMessage({ message: topicMessagePayload }));
       store.dispatch(updateTopicMessagesPhase('consuming'));
       store.dispatch(updateTopicMessagesMeta(topicMessagesMetaPayload));
     });

kafka-ui-react-app/src/redux/reducers/topicMessages/reducer.ts

@@ -1,6 +1,7 @@
 import { Action, TopicMessagesState } from 'redux/interfaces';
 import { getType } from 'typesafe-actions';
 import * as actions from 'redux/actions';
+import { TopicMessage } from 'generated-sources';
 
 export const initialState: TopicMessagesState = {
   messages: [],
@@ -17,9 +18,13 @@ export const initialState: TopicMessagesState = {
 const reducer = (state = initialState, action: Action): TopicMessagesState => {
   switch (action.type) {
     case getType(actions.addTopicMessage): {
+      const messages: TopicMessage[] = action.payload.prepend
+        ? [action.payload.message, ...state.messages]
+        : [...state.messages, action.payload.message];
+
       return {
         ...state,
-        messages: [action.payload, ...state.messages],
+        messages,
       };
     }
     case getType(actions.resetTopicMessages):