(ui): Add redux action/reducer for aggregating tasks

ui/src/actions/tasksActions.ts

@@ -34,6 +34,7 @@ import {
   listRetryTasks,
   listScheduledTasks,
   listCompletedTasks,
+  listAggregatingTasks,
   PaginationOptions,
   runAllArchivedTasks,
   runAllRetryTasks,
@@ -75,6 +76,9 @@ export const LIST_ARCHIVED_TASKS_ERROR = "LIST_ARCHIVED_TASKS_ERROR";
 export const LIST_COMPLETED_TASKS_BEGIN = "LIST_COMPLETED_TASKS_BEGIN";
 export const LIST_COMPLETED_TASKS_SUCCESS = "LIST_COMPLETED_TASKS_SUCCESS";
 export const LIST_COMPLETED_TASKS_ERROR = "LIST_COMPLETED_TASKS_ERROR";
+export const LIST_AGGREGATING_TASKS_BEGIN = "LIST_AGGREGATING_TASKS_BEGIN";
+export const LIST_AGGREGATING_TASKS_SUCCESS = "LIST_AGGREGATING_TASKS_SUCCESS";
+export const LIST_AGGREGATING_TASKS_ERROR = "LIST_AGGREGATING_TASKS_ERROR";
 export const CANCEL_ACTIVE_TASK_BEGIN = "CANCEL_ACTIVE_TASK_BEGIN";
 export const CANCEL_ACTIVE_TASK_SUCCESS = "CANCEL_ACTIVE_TASK_SUCCESS";
 export const CANCEL_ACTIVE_TASK_ERROR = "CANCEL_ACTIVE_TASK_ERROR";
@@ -348,6 +352,26 @@ interface ListCompletedTasksErrorAction {
   error: string; // error description
 }
 
+interface ListAggregatingTasksBeginAction {
+  type: typeof LIST_AGGREGATING_TASKS_BEGIN;
+  queue: string;
+  group: string;
+}
+
+interface ListAggregatingTasksSuccessAction {
+  type: typeof LIST_AGGREGATING_TASKS_SUCCESS;
+  queue: string;
+  group: string;
+  payload: ListTasksResponse;
+}
+
+interface ListAggregatingTasksErrorAction {
+  type: typeof LIST_AGGREGATING_TASKS_ERROR;
+  queue: string;
+  group: string;
+  error: string; // error description
+}
+
 interface CancelActiveTaskBeginAction {
   type: typeof CANCEL_ACTIVE_TASK_BEGIN;
   queue: string;
@@ -1024,6 +1048,9 @@ export type TasksActionTypes =
   | ListCompletedTasksBeginAction
   | ListCompletedTasksSuccessAction
   | ListCompletedTasksErrorAction
+  | ListAggregatingTasksBeginAction
+  | ListAggregatingTasksSuccessAction
+  | ListAggregatingTasksErrorAction
   | CancelActiveTaskBeginAction
   | CancelActiveTaskSuccessAction
   | CancelActiveTaskErrorAction
@@ -1314,6 +1341,40 @@ export function listCompletedTasksAsync(
   };
 }
 
+export function listAggregatingTasksAsync(
+  qname: string,
+  gname: string,
+  pageOpts?: PaginationOptions
+) {
+  return async (dispatch: Dispatch<TasksActionTypes>) => {
+    try {
+      dispatch({
+        type: LIST_AGGREGATING_TASKS_BEGIN,
+        queue: qname,
+        group: gname,
+      });
+      const response = await listAggregatingTasks(qname, gname, pageOpts);
+      dispatch({
+        type: LIST_AGGREGATING_TASKS_SUCCESS,
+        queue: qname,
+        group: gname,
+        payload: response,
+      });
+    } catch (error) {
+      console.error(
+        "listAggregatingTasksAsync: ",
+        toErrorStringWithHttpStatus(error)
+      );
+      dispatch({
+        type: LIST_AGGREGATING_TASKS_ERROR,
+        queue: qname,
+        group: gname,
+        error: toErrorString(error),
+      });
+    }
+  };
+}
+
 export function cancelActiveTaskAsync(queue: string, taskId: string) {
   return async (dispatch: Dispatch<TasksActionTypes>) => {
     dispatch({ type: CANCEL_ACTIVE_TASK_BEGIN, queue, taskId });

ui/src/api.ts

@@ -526,6 +526,22 @@ export async function listCompletedTasks(
   return resp.data;
 }
 
+export async function listAggregatingTasks(
+  qname: string,
+  gname: string,
+  pageOpts?: PaginationOptions
+): Promise<ListTasksResponse> {
+  let url = `${getBaseUrl()}/queues/${qname}/groups/${gname}/aggregating_tasks`;
+  if (pageOpts) {
+    url += `?${queryString.stringify(pageOpts)}`;
+  }
+  const resp = await axios({
+    method: "get",
+    url,
+  });
+  return resp.data;
+}
+
 export async function archivePendingTask(
   qname: string,
   taskId: string

ui/src/components/TaskGroupsTable.tsx

@@ -32,6 +32,7 @@ import { TableColumn } from "../types/table";
 import TablePaginationActions, {
   rowsPerPageOptions,
 } from "./TablePaginationActions";
+import { listAggregatingTasksAsync } from "../actions/tasksActions";
 
 const useStyles = makeStyles((theme) => ({
   groupSelector: {
@@ -60,11 +61,13 @@ function mapStateToProps(state: AppState) {
     groups: state.groups.data,
     groupsError: state.groups.error,
     pollInterval: state.settings.pollInterval,
+    pageSize: state.settings.taskRowsPerPage,
   };
 }
 
 const mapDispatchToProps = {
   listGroupsAsync,
+  listAggregatingTasksAsync,
 };
 
 const connector = connect(mapStateToProps, mapDispatchToProps);
@@ -87,15 +90,31 @@ function TaskGroupsTable(props: Props & ConnectedProps<typeof connector>) {
   const [selectedGroup, setSelectedGroup] = React.useState<GroupInfo | null>(
     null
   );
+  const [page, setPage] = React.useState(0);
   const [selectedIds, setSelectedIds] = React.useState<string[]>([]);
-  const { pollInterval, listGroupsAsync, queue } = props;
+  const [activeTaskId, setActiveTaskId] = React.useState<string>("");
+  const {
+    pollInterval,
+    listGroupsAsync,
+    listAggregatingTasksAsync,
+    queue,
+    pageSize,
+  } = props;
   const classes = useStyles();
 
   const fetchGroups = useCallback(() => {
     listGroupsAsync(queue);
   }, [listGroupsAsync, queue]);
 
+  const fetchTasks = useCallback(() => {
+    const pageOpts = { page: page + 1, size: pageSize };
+    if (selectedGroup !== null) {
+      listAggregatingTasksAsync(queue, selectedGroup.group, pageOpts);
+    }
+  }, [page, pageSize, queue, selectedGroup, listAggregatingTasksAsync]);
+
   usePolling(fetchGroups, pollInterval);
+  usePolling(fetchTasks, pollInterval);
 
   const rowCount = 0; // TODO: props.tasks.length;
   const numSelected = selectedIds.length;

ui/src/reducers/tasksReducer.ts

@@ -129,6 +129,9 @@ import {
   BATCH_DELETE_COMPLETED_TASKS_BEGIN,
   BATCH_DELETE_COMPLETED_TASKS_ERROR,
   BATCH_DELETE_COMPLETED_TASKS_SUCCESS,
+  LIST_AGGREGATING_TASKS_BEGIN,
+  LIST_AGGREGATING_TASKS_SUCCESS,
+  LIST_AGGREGATING_TASKS_ERROR,
 } from "../actions/tasksActions";
 import { TaskInfo } from "../api";
 
@@ -191,6 +194,13 @@ interface TasksState {
     error: string;
     data: TaskInfoExtended[];
   };
+  aggregatingTasks: {
+    loading: boolean;
+    batchActionPending: boolean;
+    allActionPending: boolean;
+    error: string;
+    data: TaskInfoExtended[];
+  };
   taskInfo: {
     loading: boolean;
     error: string;
@@ -241,6 +251,13 @@ const initialState: TasksState = {
     error: "",
     data: [],
   },
+  aggregatingTasks: {
+    loading: false,
+    batchActionPending: false,
+    allActionPending: false,
+    error: "",
+    data: [],
+  },
   taskInfo: {
     loading: false,
     error: "",
@@ -485,6 +502,40 @@ function tasksReducer(
         },
       };
 
+    case LIST_AGGREGATING_TASKS_BEGIN:
+      return {
+        ...state,
+        aggregatingTasks: {
+          ...state.aggregatingTasks,
+          loading: true,
+        },
+      };
+
+    case LIST_AGGREGATING_TASKS_SUCCESS:
+      return {
+        ...state,
+        aggregatingTasks: {
+          ...state.aggregatingTasks,
+          loading: false,
+          error: "",
+          data: action.payload.tasks.map((task) => ({
+            ...task,
+            requestPending: false,
+          })),
+        },
+      };
+
+    case LIST_AGGREGATING_TASKS_ERROR:
+      return {
+        ...state,
+        aggregatingTasks: {
+          ...state.aggregatingTasks,
+          loading: false,
+          error: action.error,
+          data: [],
+        },
+      };
+
     case DELETE_COMPLETED_TASK_BEGIN:
       return {
         ...state,