[GitHub] [ignite-3] valepakh commented on a diff in pull request #2359: IGNITE-19108 Check connection during REPL session in CLI

["valepakh (via GitHub)" <gi...@apache.org>]

valepakh commented on code in PR #2359:
URL: https://github.com/apache/ignite-3/pull/2359#discussion_r1286811430


##########
modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/ItConnectionHeartbeatTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.cli.commands;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import io.micronaut.context.annotation.Property;
+import io.micronaut.context.annotation.Value;
+import jakarta.inject.Inject;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.cli.core.repl.Session;
+import org.apache.ignite.internal.cli.event.EventListener;
+import org.apache.ignite.internal.cli.event.EventSubscriptionManager;
+import org.apache.ignite.internal.cli.event.EventType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+@Property(name = "cli.check.connection.period.second", value = "1")
+class ItConnectionHeartbeatTest extends CliCommandTestInitializedIntegrationBase {
+
+    @Inject
+    Session session;
+
+    @Inject
+    EventSubscriptionManager eventSubscriptionManager;
+
+    @Value("${cli.check.connection.period.second}")
+    private long cliCheckConnectionPeriodSecond;
+
+    private final AtomicInteger connectionLostCount = new AtomicInteger(0);
+    private final AtomicInteger connectionRestoredCount = new AtomicInteger(0);
+
+    @BeforeEach
+    void setUp() {
+        connectionLostCount.set(0);
+        connectionRestoredCount.set(0);
+        EventListener connectionRestoredEventListener = (event) -> {
+            connectionRestoredCount.incrementAndGet();
+        };
+
+        EventListener connectionLostEventListener = (event) -> {
+            connectionLostCount.incrementAndGet();
+        };
+
+        // Register listeners
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_LOST, connectionLostEventListener);
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_RESTORED, connectionRestoredEventListener);
+    }
+
+    @Override
+    protected Class<?> getCommandClass() {
+        return TopLevelCliReplCommand.class;
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_RESTORED on connection start")
+    void connectionEstablished() {
+        // Given null session info before connect
+        assertNull(session.info());
+
+        // When connect without parameters
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // Listener was invoked
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        assertEquals(0, connectionLostCount.get());
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_LOST on cluster stop")
+    void onConnectionLost() {
+        // Given connected cli
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // When stop node
+        String nodeName = session.info().nodeName();
+        this.stopNode(nodeName);

Review Comment:
   ```suggestion
           stopNode(nodeName);
   ```



##########
modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/ItConnectionHeartbeatTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.cli.commands;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import io.micronaut.context.annotation.Property;
+import io.micronaut.context.annotation.Value;
+import jakarta.inject.Inject;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.cli.core.repl.Session;
+import org.apache.ignite.internal.cli.event.EventListener;
+import org.apache.ignite.internal.cli.event.EventSubscriptionManager;
+import org.apache.ignite.internal.cli.event.EventType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+@Property(name = "cli.check.connection.period.second", value = "1")
+class ItConnectionHeartbeatTest extends CliCommandTestInitializedIntegrationBase {
+
+    @Inject
+    Session session;
+
+    @Inject
+    EventSubscriptionManager eventSubscriptionManager;
+
+    @Value("${cli.check.connection.period.second}")
+    private long cliCheckConnectionPeriodSecond;
+
+    private final AtomicInteger connectionLostCount = new AtomicInteger(0);
+    private final AtomicInteger connectionRestoredCount = new AtomicInteger(0);
+
+    @BeforeEach
+    void setUp() {
+        connectionLostCount.set(0);
+        connectionRestoredCount.set(0);
+        EventListener connectionRestoredEventListener = (event) -> {
+            connectionRestoredCount.incrementAndGet();
+        };
+
+        EventListener connectionLostEventListener = (event) -> {
+            connectionLostCount.incrementAndGet();
+        };
+
+        // Register listeners
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_LOST, connectionLostEventListener);
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_RESTORED, connectionRestoredEventListener);

Review Comment:
   I'd remove the local variables, or at least remove unnecessary braces and parentheses
   ```suggestion
   
           // Register listeners
           eventSubscriptionManager.subscribe(EventType.CONNECTION_LOST, event -> connectionLostCount.incrementAndGet());
           eventSubscriptionManager.subscribe(EventType.CONNECTION_RESTORED, event -> connectionRestoredCount.incrementAndGet());
   ```



##########
modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/ItConnectionHeartbeatTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.cli.commands;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import io.micronaut.context.annotation.Property;
+import io.micronaut.context.annotation.Value;
+import jakarta.inject.Inject;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.cli.core.repl.Session;
+import org.apache.ignite.internal.cli.event.EventListener;
+import org.apache.ignite.internal.cli.event.EventSubscriptionManager;
+import org.apache.ignite.internal.cli.event.EventType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+@Property(name = "cli.check.connection.period.second", value = "1")
+class ItConnectionHeartbeatTest extends CliCommandTestInitializedIntegrationBase {
+
+    @Inject
+    Session session;
+
+    @Inject
+    EventSubscriptionManager eventSubscriptionManager;
+
+    @Value("${cli.check.connection.period.second}")
+    private long cliCheckConnectionPeriodSecond;
+
+    private final AtomicInteger connectionLostCount = new AtomicInteger(0);
+    private final AtomicInteger connectionRestoredCount = new AtomicInteger(0);
+
+    @BeforeEach
+    void setUp() {
+        connectionLostCount.set(0);
+        connectionRestoredCount.set(0);
+        EventListener connectionRestoredEventListener = (event) -> {
+            connectionRestoredCount.incrementAndGet();
+        };
+
+        EventListener connectionLostEventListener = (event) -> {
+            connectionLostCount.incrementAndGet();
+        };
+
+        // Register listeners
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_LOST, connectionLostEventListener);
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_RESTORED, connectionRestoredEventListener);
+    }
+
+    @Override
+    protected Class<?> getCommandClass() {
+        return TopLevelCliReplCommand.class;
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_RESTORED on connection start")
+    void connectionEstablished() {
+        // Given null session info before connect
+        assertNull(session.info());
+
+        // When connect without parameters
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // Listener was invoked
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        assertEquals(0, connectionLostCount.get());
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_LOST on cluster stop")
+    void onConnectionLost() {
+        // Given connected cli
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // When stop node
+        String nodeName = session.info().nodeName();
+        this.stopNode(nodeName);
+
+        // Listener was invoked
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionLostCount.get() == 1);
+
+        // Tear down. Restore initial state of node to exclude any impact on next test.
+        this.startNode(nodeName);
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_LOST on cluster stop")
+    void restoreConnectionAfterConnectionLost() {
+        // Given connected cli
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // When stop node
+        String nodeName = session.info().nodeName();
+        this.stopNode(nodeName);

Review Comment:
   ```suggestion
           stopNode(nodeName);
   ```



##########
modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/ItConnectionHeartbeatTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.cli.commands;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import io.micronaut.context.annotation.Property;
+import io.micronaut.context.annotation.Value;
+import jakarta.inject.Inject;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.cli.core.repl.Session;
+import org.apache.ignite.internal.cli.event.EventListener;
+import org.apache.ignite.internal.cli.event.EventSubscriptionManager;
+import org.apache.ignite.internal.cli.event.EventType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+@Property(name = "cli.check.connection.period.second", value = "1")
+class ItConnectionHeartbeatTest extends CliCommandTestInitializedIntegrationBase {
+
+    @Inject
+    Session session;
+
+    @Inject
+    EventSubscriptionManager eventSubscriptionManager;
+
+    @Value("${cli.check.connection.period.second}")
+    private long cliCheckConnectionPeriodSecond;
+
+    private final AtomicInteger connectionLostCount = new AtomicInteger(0);
+    private final AtomicInteger connectionRestoredCount = new AtomicInteger(0);
+
+    @BeforeEach
+    void setUp() {
+        connectionLostCount.set(0);
+        connectionRestoredCount.set(0);
+        EventListener connectionRestoredEventListener = (event) -> {
+            connectionRestoredCount.incrementAndGet();
+        };
+
+        EventListener connectionLostEventListener = (event) -> {
+            connectionLostCount.incrementAndGet();
+        };
+
+        // Register listeners
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_LOST, connectionLostEventListener);
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_RESTORED, connectionRestoredEventListener);
+    }
+
+    @Override
+    protected Class<?> getCommandClass() {
+        return TopLevelCliReplCommand.class;
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_RESTORED on connection start")
+    void connectionEstablished() {
+        // Given null session info before connect
+        assertNull(session.info());
+
+        // When connect without parameters
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // Listener was invoked
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        assertEquals(0, connectionLostCount.get());
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_LOST on cluster stop")
+    void onConnectionLost() {
+        // Given connected cli
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // When stop node
+        String nodeName = session.info().nodeName();
+        this.stopNode(nodeName);
+
+        // Listener was invoked
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionLostCount.get() == 1);
+
+        // Tear down. Restore initial state of node to exclude any impact on next test.
+        this.startNode(nodeName);
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_LOST on cluster stop")
+    void restoreConnectionAfterConnectionLost() {
+        // Given connected cli
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // When stop node
+        String nodeName = session.info().nodeName();
+        this.stopNode(nodeName);
+
+        // Then connection lost event obtained
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionLostCount.get() == 1);
+
+        // When
+        this.startNode(nodeName);

Review Comment:
   ```suggestion
           startNode(nodeName);
   ```



##########
modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/ItConnectionHeartbeatTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.cli.commands;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import io.micronaut.context.annotation.Property;
+import io.micronaut.context.annotation.Value;
+import jakarta.inject.Inject;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.cli.core.repl.Session;
+import org.apache.ignite.internal.cli.event.EventListener;
+import org.apache.ignite.internal.cli.event.EventSubscriptionManager;
+import org.apache.ignite.internal.cli.event.EventType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+@Property(name = "cli.check.connection.period.second", value = "1")
+class ItConnectionHeartbeatTest extends CliCommandTestInitializedIntegrationBase {
+
+    @Inject
+    Session session;
+
+    @Inject
+    EventSubscriptionManager eventSubscriptionManager;
+
+    @Value("${cli.check.connection.period.second}")
+    private long cliCheckConnectionPeriodSecond;
+
+    private final AtomicInteger connectionLostCount = new AtomicInteger(0);
+    private final AtomicInteger connectionRestoredCount = new AtomicInteger(0);
+
+    @BeforeEach
+    void setUp() {
+        connectionLostCount.set(0);
+        connectionRestoredCount.set(0);
+        EventListener connectionRestoredEventListener = (event) -> {
+            connectionRestoredCount.incrementAndGet();
+        };
+
+        EventListener connectionLostEventListener = (event) -> {
+            connectionLostCount.incrementAndGet();
+        };
+
+        // Register listeners
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_LOST, connectionLostEventListener);
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_RESTORED, connectionRestoredEventListener);
+    }
+
+    @Override
+    protected Class<?> getCommandClass() {
+        return TopLevelCliReplCommand.class;
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_RESTORED on connection start")
+    void connectionEstablished() {
+        // Given null session info before connect
+        assertNull(session.info());
+
+        // When connect without parameters
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // Listener was invoked
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        assertEquals(0, connectionLostCount.get());
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_LOST on cluster stop")
+    void onConnectionLost() {
+        // Given connected cli
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // When stop node
+        String nodeName = session.info().nodeName();
+        this.stopNode(nodeName);
+
+        // Listener was invoked
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionLostCount.get() == 1);

Review Comment:
   I'd combine the checks:
   ```suggestion
           await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(
                   () -> connectionRestoredCount.get() == 1 && connectionLostCount.get() == 1
           );
   ```



##########
modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/ConnectionHeartBeat.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.cli.core.repl;
+
+import io.micronaut.context.annotation.Value;
+import jakarta.inject.Singleton;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import javax.annotation.Nullable;
+import org.apache.ignite.internal.cli.core.rest.ApiClientFactory;
+import org.apache.ignite.internal.cli.event.Event;
+import org.apache.ignite.internal.cli.event.EventListener;
+import org.apache.ignite.internal.cli.event.EventPublisher;
+import org.apache.ignite.internal.cli.event.EventType;
+import org.apache.ignite.internal.cli.event.Events;
+import org.apache.ignite.internal.cli.logger.CliLoggers;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.thread.NamedThreadFactory;
+import org.apache.ignite.rest.client.api.NodeManagementApi;
+import org.apache.ignite.rest.client.invoker.ApiException;
+
+/**
+ * Connection to node heart beat.
+ */
+@Singleton
+public class ConnectionHeartBeat implements EventListener {
+
+    private static final IgniteLogger log = CliLoggers.forClass(ConnectionHeartBeat.class);

Review Comment:
   Statics should be in uppercase



##########
modules/cli/src/main/java/org/apache/ignite/internal/cli/event/EventFactory.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.cli.event;
+
+import jakarta.inject.Singleton;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import org.apache.ignite.internal.cli.logger.CliLoggers;
+import org.apache.ignite.internal.logger.IgniteLogger;
+
+/**
+ * Register listeners and produces events.
+ */
+@Singleton
+public class EventFactory implements EventPublisher, EventSubscriptionManager {
+
+    private static final IgniteLogger log = CliLoggers.forClass(EventFactory.class);
+
+    /** All listeners. */
+    private final ConcurrentHashMap<EventType, List<EventListener>> listeners = new ConcurrentHashMap<>();
+
+    public EventFactory() {
+    }
+

Review Comment:
   ```suggestion
   
   ```



##########
modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/ItConnectionHeartbeatTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.cli.commands;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import io.micronaut.context.annotation.Property;
+import io.micronaut.context.annotation.Value;
+import jakarta.inject.Inject;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.cli.core.repl.Session;
+import org.apache.ignite.internal.cli.event.EventListener;
+import org.apache.ignite.internal.cli.event.EventSubscriptionManager;
+import org.apache.ignite.internal.cli.event.EventType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+@Property(name = "cli.check.connection.period.second", value = "1")
+class ItConnectionHeartbeatTest extends CliCommandTestInitializedIntegrationBase {
+
+    @Inject
+    Session session;
+
+    @Inject
+    EventSubscriptionManager eventSubscriptionManager;
+
+    @Value("${cli.check.connection.period.second}")
+    private long cliCheckConnectionPeriodSecond;
+
+    private final AtomicInteger connectionLostCount = new AtomicInteger(0);
+    private final AtomicInteger connectionRestoredCount = new AtomicInteger(0);
+
+    @BeforeEach
+    void setUp() {
+        connectionLostCount.set(0);
+        connectionRestoredCount.set(0);
+        EventListener connectionRestoredEventListener = (event) -> {
+            connectionRestoredCount.incrementAndGet();
+        };
+
+        EventListener connectionLostEventListener = (event) -> {
+            connectionLostCount.incrementAndGet();
+        };
+
+        // Register listeners
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_LOST, connectionLostEventListener);
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_RESTORED, connectionRestoredEventListener);
+    }
+
+    @Override
+    protected Class<?> getCommandClass() {
+        return TopLevelCliReplCommand.class;
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_RESTORED on connection start")
+    void connectionEstablished() {
+        // Given null session info before connect
+        assertNull(session.info());
+
+        // When connect without parameters
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // Listener was invoked
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        assertEquals(0, connectionLostCount.get());
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_LOST on cluster stop")
+    void onConnectionLost() {
+        // Given connected cli
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // When stop node
+        String nodeName = session.info().nodeName();
+        this.stopNode(nodeName);
+
+        // Listener was invoked
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionLostCount.get() == 1);
+
+        // Tear down. Restore initial state of node to exclude any impact on next test.
+        this.startNode(nodeName);

Review Comment:
   ```suggestion
           startNode(nodeName);
   ```



##########
modules/cli/src/main/java/org/apache/ignite/internal/cli/event/EventFactory.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.cli.event;
+
+import jakarta.inject.Singleton;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import org.apache.ignite.internal.cli.logger.CliLoggers;
+import org.apache.ignite.internal.logger.IgniteLogger;
+
+/**
+ * Register listeners and produces events.
+ */
+@Singleton
+public class EventFactory implements EventPublisher, EventSubscriptionManager {
+
+    private static final IgniteLogger log = CliLoggers.forClass(EventFactory.class);

Review Comment:
   LOG



##########
modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/ItConnectionHeartbeatTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.cli.commands;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import io.micronaut.context.annotation.Property;
+import io.micronaut.context.annotation.Value;
+import jakarta.inject.Inject;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.cli.core.repl.Session;
+import org.apache.ignite.internal.cli.event.EventListener;
+import org.apache.ignite.internal.cli.event.EventSubscriptionManager;
+import org.apache.ignite.internal.cli.event.EventType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+@Property(name = "cli.check.connection.period.second", value = "1")
+class ItConnectionHeartbeatTest extends CliCommandTestInitializedIntegrationBase {
+
+    @Inject
+    Session session;
+
+    @Inject
+    EventSubscriptionManager eventSubscriptionManager;
+
+    @Value("${cli.check.connection.period.second}")
+    private long cliCheckConnectionPeriodSecond;
+
+    private final AtomicInteger connectionLostCount = new AtomicInteger(0);
+    private final AtomicInteger connectionRestoredCount = new AtomicInteger(0);
+
+    @BeforeEach
+    void setUp() {
+        connectionLostCount.set(0);
+        connectionRestoredCount.set(0);
+        EventListener connectionRestoredEventListener = (event) -> {
+            connectionRestoredCount.incrementAndGet();
+        };
+
+        EventListener connectionLostEventListener = (event) -> {
+            connectionLostCount.incrementAndGet();
+        };
+
+        // Register listeners
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_LOST, connectionLostEventListener);
+        eventSubscriptionManager.subscribe(EventType.CONNECTION_RESTORED, connectionRestoredEventListener);
+    }
+
+    @Override
+    protected Class<?> getCommandClass() {
+        return TopLevelCliReplCommand.class;
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_RESTORED on connection start")
+    void connectionEstablished() {
+        // Given null session info before connect
+        assertNull(session.info());
+
+        // When connect without parameters
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // Listener was invoked
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        assertEquals(0, connectionLostCount.get());
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_LOST on cluster stop")
+    void onConnectionLost() {
+        // Given connected cli
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // When stop node
+        String nodeName = session.info().nodeName();
+        this.stopNode(nodeName);
+
+        // Listener was invoked
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionLostCount.get() == 1);
+
+        // Tear down. Restore initial state of node to exclude any impact on next test.
+        this.startNode(nodeName);
+    }
+
+    @Test
+    @DisplayName("Should send event CONNECTION_LOST on cluster stop")
+    void restoreConnectionAfterConnectionLost() {
+        // Given connected cli
+        execute("connect");
+
+        // Then
+        assertAll(
+                this::assertErrOutputIsEmpty,
+                () -> assertOutputContains("Connected to http://localhost:10300")
+        );
+
+        // When stop node
+        String nodeName = session.info().nodeName();
+        this.stopNode(nodeName);
+
+        // Then connection lost event obtained
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionRestoredCount.get() == 1);
+        await().timeout(cliCheckConnectionPeriodSecond * 2, TimeUnit.SECONDS).until(() -> connectionLostCount.get() == 1);

Review Comment:
   And here and below as well.



##########
modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/impl/ClusterConfigRegistryImpl.java:
##########
@@ -24,26 +24,35 @@
 import org.apache.ignite.internal.cli.call.configuration.ClusterConfigShowCallInput;
 import org.apache.ignite.internal.cli.call.configuration.JsonString;
 import org.apache.ignite.internal.cli.core.call.DefaultCallOutput;
-import org.apache.ignite.internal.cli.core.repl.AsyncSessionEventListener;
+import org.apache.ignite.internal.cli.core.repl.ConnectEvent;
 import org.apache.ignite.internal.cli.core.repl.SessionInfo;
 import org.apache.ignite.internal.cli.core.repl.registry.ClusterConfigRegistry;
+import org.apache.ignite.internal.cli.event.Event;
+import org.apache.ignite.internal.cli.event.EventListener;
+import org.apache.ignite.internal.cli.event.EventType;
 import org.jetbrains.annotations.Nullable;
 
 /** Implementation of {@link ClusterConfigRegistry}. */
 @Singleton
-public class ClusterConfigRegistryImpl implements ClusterConfigRegistry, AsyncSessionEventListener {
+public class ClusterConfigRegistryImpl implements ClusterConfigRegistry, EventListener {
 
     private final ClusterConfigShowCall clusterConfigShowCall;
 
+    @Nullable
     private LazyObjectRef<Config> configRef;
 
     public ClusterConfigRegistryImpl(ClusterConfigShowCall clusterConfigShowCall) {
         this.clusterConfigShowCall = clusterConfigShowCall;
     }
 
     @Override
-    public void onConnect(SessionInfo sessionInfo) {
-        configRef = new LazyObjectRef<>(() -> fetchConfig(sessionInfo));
+    public void onEvent(Event event) {
+        if (EventType.CONNECT == event.eventType()) {
+            ConnectEvent connectEvent = (ConnectEvent) event;
+            configRef = new LazyObjectRef<>(() -> fetchConfig(connectEvent.sessionInfo()));
+        } else if (EventType.DISCONNECT == event.eventType()) {
+            configRef = null;
+        }
     }

Review Comment:
   Seeing this pattern many times I wonder whether it's worth changing `onEvent` to be default method and adding `onConnect` and `onDisconnect` default methods to the `EventListener`?
   This will keep the current code intact.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org