Fix live streaming

x-pack/legacy/plugins/infra/public/containers/logs/log_entries/index.ts

@@ -5,7 +5,7 @@
  */
 import { useEffect, useState, useReducer, useCallback } from 'react';
 import createContainer from 'constate';
-import { pick, throttle } from 'lodash';
+import { pick, throttle, omit } from 'lodash';
 import { useGraphQLQueries } from './gql_queries';
 import { TimeKey, timeKeyIsBetween } from '../../../../common/time';
 import { InfraLogEntry } from './types';
@@ -45,6 +45,7 @@ interface LogEntriesProps {
   pagesAfterEnd: number | null;
   sourceId: string;
   isAutoReloading: boolean;
+  jumpToTargetPosition: (position: TimeKey) => void;
 }
 
 type FetchEntriesParams = Omit<LogEntriesProps, 'isAutoReloading'>;
@@ -65,7 +66,7 @@ export type LogEntriesStateParams = {
 } & LogEntriesResponse;
 
 export interface LogEntriesCallbacks {
-  fetchNewerEntries: () => Promise<void>;
+  fetchNewerEntries: () => Promise<TimeKey | null | undefined>;
 }
 export const logEntriesInitialCallbacks = {
   fetchNewerEntries: async () => {},
@@ -127,10 +128,13 @@ const useFetchEntriesEffect = (
   const [prevParams, cachePrevParams] = useState(props);
   const [startedStreaming, setStartedStreaming] = useState(false);
 
-  const runFetchNewEntriesRequest = async () => {
+  const runFetchNewEntriesRequest = async (override = {}) => {
     dispatch({ type: Action.FetchingNewEntries });
     try {
-      const payload = await getLogEntriesAround(props);
+      const payload = await getLogEntriesAround({
+        ...omit(props, 'jumpToTargetPosition'),
+        ...override,
+      });
       dispatch({ type: Action.ReceiveNewEntries, payload });
     } catch (e) {
       dispatch({ type: Action.ErrorOnNewEntries });
@@ -150,6 +154,7 @@ const useFetchEntriesEffect = (
         type: getEntriesBefore ? Action.ReceiveEntriesBefore : Action.ReceiveEntriesAfter,
         payload,
       });
+      return payload.entriesEnd;
     } catch (e) {
       dispatch({ type: Action.ErrorOnMoreEntries });
     }
@@ -185,19 +190,37 @@ const useFetchEntriesEffect = (
 
   const fetchNewerEntries = useCallback(
     throttle(() => runFetchMoreEntriesRequest(ShouldFetchMoreEntries.After), 500),
-    [props]
+    [props, state.entriesEnd]
   );
 
-  const streamEntriesEffectDependencies = [props.isAutoReloading, state.isLoadingMore];
+  const streamEntriesEffectDependencies = [
+    props.isAutoReloading,
+    state.isLoadingMore,
+    state.isReloading,
+  ];
   const streamEntriesEffect = () => {
     (async () => {
-      if (props.isAutoReloading && !state.isLoadingMore) {
+      if (props.isAutoReloading && !state.isLoadingMore && !state.isReloading) {
         if (startedStreaming) {
           await new Promise(res => setTimeout(res, 5000));
         } else {
+          const nowKey = {
+            tiebreaker: 0,
+            time: Date.now(),
+          };
+          props.jumpToTargetPosition(nowKey);
           setStartedStreaming(true);
+          if (state.hasMoreAfterEnd) {
+            runFetchNewEntriesRequest({
+              timeKey: nowKey,
+            });
+            return;
+          }
+        }
+        const newEntriesEnd = await runFetchMoreEntriesRequest(ShouldFetchMoreEntries.After);
+        if (newEntriesEnd) {
+          props.jumpToTargetPosition(newEntriesEnd);
         }
-        fetchNewerEntries();
       } else if (!props.isAutoReloading) {
         setStartedStreaming(false);
       }

x-pack/legacy/plugins/infra/public/containers/logs/log_position/log_position_state.ts

@@ -31,7 +31,7 @@ export interface LogPositionStateParams {
 
 export interface LogPositionCallbacks {
   jumpToTargetPosition: (pos: TimeKeyOrNull) => void;
-  jumpToTargetPositionTime: (time: number, fromAutoReload?: boolean) => void;
+  jumpToTargetPositionTime: (time: number) => void;
   reportVisiblePositions: (visPos: VisiblePositions) => void;
   startLiveStreaming: () => void;
   stopLiveStreaming: () => void;
@@ -92,10 +92,9 @@ export const useLogPositionState: () => [LogPositionStateParams, LogPositionCall
 
   const callbacks = {
     jumpToTargetPosition,
-    jumpToTargetPositionTime: (time: number, fromAutoReload: boolean = false) =>
-      jumpToTargetPosition({ tiebreaker: 0, time, fromAutoReload }),
+    jumpToTargetPositionTime: (time: number) => jumpToTargetPosition({ tiebreaker: 0, time }),
     reportVisiblePositions,
-    startLiveStreaming: () => callbacks.jumpToTargetPositionTime(Date.now()),
+    startLiveStreaming: () => setIsAutoReloading(true),
     stopLiveStreaming: () => setIsAutoReloading(false),
   };
 

x-pack/legacy/plugins/infra/public/containers/logs/log_summary/with_summary.ts

@@ -5,40 +5,34 @@
  */
 
 import { useContext } from 'react';
-import { connect } from 'react-redux';
 
-import { logPositionSelectors, State } from '../../../store';
 import { RendererFunction } from '../../../utils/typed_react';
 import { Source } from '../../source';
 import { LogViewConfiguration } from '../log_view_configuration';
 import { LogSummaryBuckets, useLogSummary } from './log_summary';
 import { LogFilterState } from '../log_filter';
+import { LogPositionState } from '../log_position';
 
-export const WithSummary = connect((state: State) => ({
-  visibleMidpointTime: logPositionSelectors.selectVisibleMidpointOrTargetTime(state),
-}))(
-  ({
-    children,
-    visibleMidpointTime,
-  }: {
-    children: RendererFunction<{
-      buckets: LogSummaryBuckets;
-      start: number | null;
-      end: number | null;
-    }>;
-    visibleMidpointTime: number | null;
-  }) => {
-    const { intervalSize } = useContext(LogViewConfiguration.Context);
-    const { sourceId } = useContext(Source.Context);
-    const { filterQuery } = useContext(LogFilterState.Context);
+export const WithSummary = ({
+  children,
+}: {
+  children: RendererFunction<{
+    buckets: LogSummaryBuckets;
+    start: number | null;
+    end: number | null;
+  }>;
+}) => {
+  const { intervalSize } = useContext(LogViewConfiguration.Context);
+  const { sourceId } = useContext(Source.Context);
+  const { filterQuery } = useContext(LogFilterState.Context);
+  const [{ visibleMidpointTime }] = useContext(LogPositionState.Context);
 
-    const { buckets, start, end } = useLogSummary(
-      sourceId,
-      visibleMidpointTime,
-      intervalSize,
-      filterQuery
-    );
+  const { buckets, start, end } = useLogSummary(
+    sourceId,
+    visibleMidpointTime,
+    intervalSize,
+    filterQuery
+  );
 
-    return children({ buckets, start, end });
-  }
-);
+  return children({ buckets, start, end });
+};

x-pack/legacy/plugins/infra/public/containers/logs/with_stream_items.ts

@@ -24,18 +24,17 @@ export const WithStreamItems: React.FunctionComponent<{
   >;
 }> = ({ children }) => {
   const [logEntries, logEntriesCallbacks] = useContext(LogEntriesState.Context);
-  const { isAutoReloading } = useContext(LogPositionState.Context);
   const { currentHighlightKey, logEntryHighlightsById } = useContext(LogHighlightsState.Context);
 
   const items = useMemo(
     () =>
-      logEntries.isReloading && !isAutoReloading
+      logEntries.isReloading
         ? []
         : logEntries.entries.map(logEntry =>
             createLogEntryStreamItem(logEntry, logEntryHighlightsById[logEntry.gid] || [])
           ),
 
-    [isAutoReloading, logEntries.entries, logEntries.isReloading, logEntryHighlightsById]
+    [logEntries.entries, logEntries.isReloading, logEntryHighlightsById]
   );
 
   return children({

x-pack/legacy/plugins/infra/public/pages/logs/stream/page_providers.tsx

@@ -4,7 +4,7 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import React, { useContext } from 'react';
+import React, { useContext, useCallback } from 'react';
 
 import { LogFlyout } from '../../../containers/logs/log_flyout';
 import { LogViewConfiguration } from '../../../containers/logs/log_view_configuration';
@@ -25,17 +25,20 @@ const LogFilterStateProvider: React.FC = ({ children }) => {
 
 const LogEntriesStateProvider: React.FC = ({ children }) => {
   const { sourceId } = useContext(Source.Context);
-  const [{ targetPosition, pagesBeforeStart, pagesAfterEnd, isAutoReloading }] = useContext(
-    LogPositionState.Context
-  );
+  const [
+    { targetPosition, pagesBeforeStart, pagesAfterEnd, isAutoReloading },
+    { jumpToTargetPosition },
+  ] = useContext(LogPositionState.Context);
   const { filterQuery } = useContext(LogFilterState.Context);
+
   const entriesProps = {
     timeKey: targetPosition,
     pagesBeforeStart,
     pagesAfterEnd,
     filterQuery,
     sourceId,
     isAutoReloading,
+    jumpToTargetPosition,
   };
   return <LogEntriesState.Provider {...entriesProps}>{children}</LogEntriesState.Provider>;
 };

x-pack/legacy/plugins/infra/public/store/actions.ts

@@ -4,9 +4,4 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-export {
-  logPositionActions,
-  waffleFilterActions,
-  waffleTimeActions,
-  waffleOptionsActions,
-} from './local';
+export { waffleFilterActions, waffleTimeActions, waffleOptionsActions } from './local';

x-pack/legacy/plugins/infra/public/store/local/actions.ts

@@ -4,7 +4,6 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-export { logPositionActions } from './log_position';
 export { waffleFilterActions } from './waffle_filter';
 export { waffleTimeActions } from './waffle_time';
 export { waffleOptionsActions } from './waffle_options';