feat(multiplayer): timeout for cato main net connection.

HMCL/src/main/java/org/jackhuang/hmcl/ui/download/VersionsPage.java

@@ -48,6 +48,7 @@ import org.jackhuang.hmcl.ui.construct.RipplerContainer;
 import org.jackhuang.hmcl.ui.wizard.Navigation;
 import org.jackhuang.hmcl.ui.wizard.Refreshable;
 import org.jackhuang.hmcl.ui.wizard.WizardPage;
+import org.jackhuang.hmcl.util.HMCLService;
 import org.jackhuang.hmcl.util.StringUtils;
 import org.jackhuang.hmcl.util.i18n.Locales;
 
@@ -279,6 +280,6 @@ public final class VersionsPage extends BorderPane implements WizardPage, Refres
 
     @FXML
     private void onSponsor() {
-        FXUtils.openLink("https://hmcl.huangyuhui.net/api/redirect/bmclapi_sponsor");
+        HMCLService.openRedirectLink("bmclapi_sponsor");
     }
 }

HMCL/src/main/java/org/jackhuang/hmcl/ui/multiplayer/MultiplayerManager.java

@@ -35,7 +35,10 @@ import org.jackhuang.hmcl.util.platform.ManagedProcess;
 import org.jackhuang.hmcl.util.platform.OperatingSystem;
 import org.jetbrains.annotations.Nullable;
 
-import java.io.*;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.io.UncheckedIOException;
 import java.net.ServerSocket;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
@@ -44,6 +47,7 @@ import java.nio.file.attribute.PosixFilePermission;
 import java.util.*;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
 import java.util.logging.Level;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -120,6 +124,19 @@ public final class MultiplayerManager {
 
             BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(process.getOutputStream(), StandardCharsets.UTF_8));
 
+            Consumer<CatoExitEvent> onExit = event -> {
+                boolean ready = session.isReady();
+                switch (event.getExitCode()) {
+                    case 1:
+                        if (!ready) {
+                            future.completeExceptionally(new CatoExitTimeoutException());
+                        }
+                        break;
+                }
+                future.completeExceptionally(new CatoExitException(event.getExitCode(), ready));
+            };
+            session.onExit.register(onExit);
+
             session.onExit().register(() -> {
                 try {
                     writer.close();
@@ -128,7 +145,14 @@ public final class MultiplayerManager {
                 }
             });
 
+            TimerTask peerConnectionTimeoutTask = Lang.setTimeout(() -> {
+                future.completeExceptionally(new PeerConnectionTimeoutException());
+                session.stop();
+            }, 15 * 1000);
+
             session.onPeerConnected.register(event -> {
+                peerConnectionTimeoutTask.cancel();
+
                 MultiplayerClient client = new MultiplayerClient(session.getId(), localPort);
                 session.addRelatedThread(client);
                 session.setClient(client);
@@ -148,6 +172,7 @@ public final class MultiplayerManager {
                         writer.write(command);
                         writer.newLine();
                         writer.flush();
+                        session.onExit.unregister(onExit);
                         future.complete(session);
                     } catch (IOException e) {
                         future.completeExceptionally(e);
@@ -178,32 +203,72 @@ public final class MultiplayerManager {
         });
     }
 
-    public static CatoSession createSession(String token, String sessionName, int gamePort, boolean allowAllJoinRequests) throws IOException {
+    public static CompletableFuture<CatoSession> createSession(String token, String sessionName, int gamePort, boolean allowAllJoinRequests) {
-        Path exe = getCatoExecutable();
+        return CompletableFuture.completedFuture(null).thenComposeAsync(unused -> {
-        if (!Files.isRegularFile(exe)) {
+            Path exe = getCatoExecutable();
-            throw new FileNotFoundException("Cato file not found");
+            if (!Files.isRegularFile(exe)) {
-        }
+                throw new CatoNotExistsException(exe);
+            }
 
-        if (!isPortAvailable(3478)) {
+            if (!isPortAvailable(3478)) {
-            throw new CatoAlreadyStartedException();
+                throw new CatoAlreadyStartedException();
-        }
+            }
 
-        LOG.info(String.format("Creating session (token=%s,sessionName=%s,gamePort=%d)", token, sessionName, gamePort));
+            LOG.info(String.format("Creating session (token=%s,sessionName=%s,gamePort=%d)", token, sessionName, gamePort));
 
-        MultiplayerServer server = new MultiplayerServer(gamePort, allowAllJoinRequests);
+            MultiplayerServer server;
-        server.startServer();
+            try {
+                server = new MultiplayerServer(gamePort, allowAllJoinRequests);
+                server.startServer();
+            } catch (IOException e) {
+                throw new UncheckedIOException(e);
+            }
 
-        String[] commands = new String[]{exe.toString(),
+            String[] commands = new String[]{exe.toString(),
-                "--token", StringUtils.isBlank(token) ? "new" : token,
+                    "--token", StringUtils.isBlank(token) ? "new" : token,
-                "--allows", String.format("%s:%d/%s:%d", REMOTE_ADDRESS, server.getPort(), REMOTE_ADDRESS, gamePort)};
+                    "--allows", String.format("%s:%d/%s:%d", REMOTE_ADDRESS, server.getPort(), REMOTE_ADDRESS, gamePort)};
-        Process process = new ProcessBuilder()
+            Process process;
-                .command(commands)
+            try {
-                .start();
+                process = new ProcessBuilder()
+                        .command(commands)
+                        .start();
+            } catch (IOException e) {
+                throw new UncheckedIOException(e);
+            }
 
-        CatoSession session = new CatoSession(sessionName, State.MASTER, process, Arrays.asList(commands));
+            CompletableFuture<CatoSession> future = new CompletableFuture<>();
-        session.setServer(server);
+
-        session.addRelatedThread(server);
+            CatoSession session = new CatoSession(sessionName, State.MASTER, process, Arrays.asList(commands));
-        return session;
+
+            Consumer<CatoExitEvent> onExit = event -> {
+                boolean ready = session.isReady();
+                switch (event.getExitCode()) {
+                    case 1:
+                        if (!ready) {
+                            future.completeExceptionally(new CatoExitTimeoutException());
+                        }
+                        break;
+                }
+                future.completeExceptionally(new CatoExitException(event.getExitCode(), ready));
+            };
+
+            session.onExit.register(onExit);
+            session.setServer(server);
+            session.addRelatedThread(server);
+
+            TimerTask peerConnectionTimeoutTask = Lang.setTimeout(() -> {
+                future.completeExceptionally(new PeerConnectionTimeoutException());
+                session.stop();
+            }, 15 * 1000);
+
+            session.onPeerConnected.register(event -> {
+                session.onExit.unregister(onExit);
+                future.complete(session);
+                peerConnectionTimeoutTask.cancel();
+            });
+
+            return future;
+        });
     }
 
     public static Invitation parseInvitationCode(String invitationCode) throws JsonParseException {
@@ -471,12 +536,39 @@ public final class MultiplayerManager {
         }
     }
 
+    public static class CatoExitException extends RuntimeException {
+        private final int exitCode;
+        private final boolean ready;
+
+        public CatoExitException(int exitCode, boolean ready) {
+            this.exitCode = exitCode;
+            this.ready = ready;
+        }
+
+        public int getExitCode() {
+            return exitCode;
+        }
+
+        public boolean isReady() {
+            return ready;
+        }
+    }
+
+    public static class CatoExitTimeoutException extends RuntimeException {
+    }
+
+    public static class CatoSessionExpiredException extends RuntimeException {
+    }
+
     public static class CatoAlreadyStartedException extends RuntimeException {
     }
 
     public static class JoinRequestTimeoutException extends RuntimeException {
     }
 
+    public static class PeerConnectionTimeoutException extends RuntimeException {
+    }
+
     public static class ConnectionErrorException extends RuntimeException {
     }
 

HMCL/src/main/java/org/jackhuang/hmcl/ui/multiplayer/MultiplayerPage.java

@@ -37,11 +37,13 @@ import org.jackhuang.hmcl.ui.FXUtils;
 import org.jackhuang.hmcl.ui.construct.*;
 import org.jackhuang.hmcl.ui.decorator.DecoratorAnimatedPage;
 import org.jackhuang.hmcl.ui.decorator.DecoratorPage;
+import org.jackhuang.hmcl.util.HMCLService;
 import org.jackhuang.hmcl.util.StringUtils;
 
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import java.util.logging.Level;
 
 import static org.jackhuang.hmcl.setting.ConfigHolder.globalConfig;
@@ -146,7 +148,7 @@ public class MultiplayerPage extends DecoratorAnimatedPage implements DecoratorP
             agreementPane.setHeading(new Label(i18n("launcher.agreement")));
             agreementPane.setBody(new Label(i18n("multiplayer.agreement.prompt")));
             JFXHyperlink agreementLink = new JFXHyperlink(i18n("launcher.agreement"));
-            agreementLink.setOnAction(e -> FXUtils.openLink("https://noin.cn/agreement"));
+            agreementLink.setOnAction(e -> HMCLService.openRedirectLink("multiplayer-agreement"));
             JFXButton yesButton = new JFXButton(i18n("launcher.agreement.accept"));
             yesButton.getStyleClass().add("dialog-accept");
             yesButton.setOnAction(e -> {
@@ -213,48 +215,36 @@ public class MultiplayerPage extends DecoratorAnimatedPage implements DecoratorP
         Controllers.dialog(new CreateMultiplayerRoomDialog((result, resolve, reject) -> {
             int gamePort = result.getServer().getAd();
             boolean isStaticToken = StringUtils.isNotBlank(globalConfig().getMultiplayerToken());
-            try {
+            MultiplayerManager.createSession(globalConfig().getMultiplayerToken(), result.getServer().getMotd(), gamePort, result.isAllowAllJoinRequests())
-                MultiplayerManager.CatoSession session = MultiplayerManager.createSession(globalConfig().getMultiplayerToken(), result.getServer().getMotd(), gamePort, result.isAllowAllJoinRequests());
+                    .thenAcceptAsync(session -> {
-                session.getServer().setOnClientAdding((client, resolveClient, rejectClient) -> {
+                        session.getServer().setOnClientAdding((client, resolveClient, rejectClient) -> {
-                    runInFX(() -> {
+                            runInFX(() -> {
-                        Controllers.dialog(new MessageDialogPane.Builder(i18n("multiplayer.session.create.join.prompt", client.getUsername()), i18n("multiplayer.session.create.join"), MessageDialogPane.MessageType.INFO)
+                                Controllers.dialog(new MessageDialogPane.Builder(i18n("multiplayer.session.create.join.prompt", client.getUsername()), i18n("multiplayer.session.create.join"), MessageDialogPane.MessageType.INFO)
-                                .yesOrNo(resolveClient, () -> rejectClient.accept(i18n("multiplayer.session.join.wait_timeout")))
+                                        .yesOrNo(resolveClient, () -> rejectClient.accept(i18n("multiplayer.session.join.wait_timeout")))
-                                .cancelOnTimeout(30 * 1000)
+                                        .cancelOnTimeout(30 * 1000)
-                                .build());
+                                        .build());
-                    });
+                            });
-                });
+                        });
-                session.getServer().onClientAdded().register(event -> {
+                        session.getServer().onClientAdded().register(event -> {
-                    runInFX(() -> {
+                            runInFX(() -> {
-                        clients.add(event);
+                                clients.add(event);
-                    });
+                            });
-                });
+                        });
-                session.getServer().onClientDisconnected().register(event -> {
+                        session.getServer().onClientDisconnected().register(event -> {
-                    runInFX(() -> {
+                            runInFX(() -> {
-                        clients.remove(event);
+                                clients.remove(event);
-                    });
+                            });
-                });
+                        });
-                initCatoSession(session);
+                        initCatoSession(session);
-            } catch (MultiplayerManager.CatoAlreadyStartedException e) {
-                LOG.log(Level.WARNING, "Cato already started", e);
-                reject.accept(i18n("multiplayer.session.error.already_started"));
-                return;
-            } catch (MultiplayerManager.CatoNotExistsException e) {
-                LOG.log(Level.WARNING, "Cato not found " + e.getFile(), e);
-                reject.accept(i18n("multiplayer.session.error.file_not_found"));
-                return;
-            } catch (Exception e) {
-                LOG.log(Level.WARNING, "Failed to create session", e);
-                if (isStaticToken) {
-                    reject.accept(i18n("multiplayer.session.create.error.static_token") + e.getLocalizedMessage());
-                } else {
-                    reject.accept(i18n("multiplayer.session.create.error.dynamic_token") + e.getLocalizedMessage());
-                }
-                return;
-            }
 
-            this.gamePort.set(gamePort);
+                        this.gamePort.set(gamePort);
-            setMultiplayerState(MultiplayerManager.State.CONNECTING);
+                        setMultiplayerState(MultiplayerManager.State.CONNECTING);
-            resolve.run();
+                        resolve.run();
+                    })
+                    .exceptionally(throwable -> {
+                        reject.accept(localizeCreateErrorMessage(throwable, isStaticToken));
+                        return null;
+                    });
         }));
     }
 
@@ -328,31 +318,7 @@ public class MultiplayerPage extends DecoratorAnimatedPage implements DecoratorP
                             resolve.run();
                         }, Platform::runLater)
                         .exceptionally(throwable -> {
-                            Throwable resolved = resolveException(throwable);
+                            reject.accept(localizeJoinErrorMessage(throwable));
-                            if (resolved instanceof CancellationException) {
-                                LOG.info("Connection rejected by the server");
-                                reject.accept(i18n("multiplayer.session.join.rejected"));
-                                return null;
-                            } else if (resolved instanceof MultiplayerManager.CatoAlreadyStartedException) {
-                                LOG.info("Cato already started");
-                                reject.accept(i18n("multiplayer.session.error.already_started"));
-                                return null;
-                            } else if (throwable instanceof MultiplayerManager.CatoNotExistsException) {
-                                LOG.log(Level.WARNING, "Cato not found " + ((MultiplayerManager.CatoNotExistsException) throwable).getFile(), throwable);
-                                reject.accept(i18n("multiplayer.session.error.file_not_found"));
-                                return null;
-                            } else if (resolved instanceof MultiplayerManager.JoinRequestTimeoutException) {
-                                LOG.info("Cato already started");
-                                reject.accept(i18n("multiplayer.session.join.wait_timeout"));
-                                return null;
-                            } else if (resolved instanceof MultiplayerManager.ConnectionErrorException) {
-                                LOG.info("Failed to establish connection with server");
-                                reject.accept(i18n("multiplayer.session.join.error.connection"));
-                                return null;
-                            } else {
-                                LOG.log(Level.WARNING, "Failed to join session", resolved);
-                                reject.accept(i18n("multiplayer.session.join.error"));
-                            }
                             return null;
                         });
             } catch (MultiplayerManager.IncompatibleCatoVersionException e) {
@@ -363,6 +329,56 @@ public class MultiplayerPage extends DecoratorAnimatedPage implements DecoratorP
                 .addQuestion(new PromptDialogPane.Builder.StringQuestion(i18n("multiplayer.session.join.invitation_code"), "", new RequiredValidator())));
     }
 
+    private String localizeErrorMessage(Throwable t, Function<Throwable, String> fallback) {
+        Throwable e = resolveException(t);
+        if (e instanceof CancellationException) {
+            LOG.info("Connection rejected by the server");
+            return i18n("multiplayer.session.join.rejected");
+        } else if (e instanceof MultiplayerManager.CatoAlreadyStartedException) {
+            LOG.info("Cato already started");
+            return i18n("multiplayer.session.error.already_started");
+        } else if (e instanceof MultiplayerManager.CatoNotExistsException) {
+            LOG.log(Level.WARNING, "Cato not found " + ((MultiplayerManager.CatoNotExistsException) e).getFile(), e);
+            return i18n("multiplayer.session.error.file_not_found");
+        } else if (e instanceof MultiplayerManager.JoinRequestTimeoutException) {
+            LOG.info("Cato already started");
+            return i18n("multiplayer.session.join.wait_timeout");
+        } else if (e instanceof MultiplayerManager.ConnectionErrorException) {
+            LOG.info("Failed to establish connection with server");
+            return i18n("multiplayer.session.join.error.connection");
+        } else if (e instanceof MultiplayerManager.CatoExitTimeoutException) {
+            LOG.info("Cato failed to connect to main net");
+            return i18n("multiplayer.exit.timeout");
+        } else if (e instanceof MultiplayerManager.CatoExitException) {
+            LOG.info("Cato exited accidentally");
+            if (!((MultiplayerManager.CatoExitException) e).isReady()) {
+                return i18n("multiplayer.exit.before_ready");
+            } else {
+                return i18n("multiplayer.exit.after_ready");
+            }
+        } else {
+            return fallback.apply(e);
+        }
+    }
+
+    private String localizeCreateErrorMessage(Throwable t, boolean isStaticToken) {
+        return localizeErrorMessage(t, e -> {
+            LOG.log(Level.WARNING, "Failed to create session", e);
+            if (isStaticToken) {
+                return i18n("multiplayer.session.create.error.static_token") + e.getLocalizedMessage();
+            } else {
+                return i18n("multiplayer.session.create.error.dynamic_token") + e.getLocalizedMessage();
+            }
+        });
+    }
+
+    private String localizeJoinErrorMessage(Throwable t) {
+        return localizeErrorMessage(t, e -> {
+            LOG.log(Level.WARNING, "Failed to join session", e);
+            return i18n("multiplayer.session.join.error");
+        });
+    }
+
     public void kickPlayer(MultiplayerChannel.CatoClient client) {
         if (getSession() == null || !getSession().isReady() || getMultiplayerState() != MultiplayerManager.State.MASTER) {
             throw new IllegalStateException("CatoSession not ready");

HMCL/src/main/java/org/jackhuang/hmcl/ui/multiplayer/MultiplayerPageSkin.java

@@ -28,7 +28,6 @@ import javafx.scene.Node;
 import javafx.scene.control.Label;
 import javafx.scene.control.ScrollPane;
 import javafx.scene.layout.*;
-import org.jackhuang.hmcl.Metadata;
 import org.jackhuang.hmcl.game.LauncherHelper;
 import org.jackhuang.hmcl.setting.Profile;
 import org.jackhuang.hmcl.setting.Profiles;
@@ -41,6 +40,7 @@ import org.jackhuang.hmcl.ui.animation.TransitionPane;
 import org.jackhuang.hmcl.ui.construct.*;
 import org.jackhuang.hmcl.ui.decorator.DecoratorAnimatedPage;
 import org.jackhuang.hmcl.ui.versions.Versions;
+import org.jackhuang.hmcl.util.HMCLService;
 import org.jackhuang.hmcl.util.javafx.BindingMapping;
 import org.jackhuang.hmcl.util.javafx.MappedObservableList;
 
@@ -130,9 +130,9 @@ public class MultiplayerPageSkin extends DecoratorAnimatedPage.DecoratorAnimated
                         item.setOnAction(e -> FXUtils.openLink("https://hmcl.huangyuhui.net/help/launcher/multiplayer.html"));
                     })
                     .addNavigationDrawerItem(report -> {
-                        report.setTitle(i18n("multiplayer.report"));
+                        report.setTitle(i18n("feedback"));
-                        report.setLeftGraphic(wrap(SVG::bug));
+                        report.setLeftGraphic(wrap(SVG::messageAlertOutline));
-                        report.setOnAction(e -> FXUtils.openLink(Metadata.EULA_URL));
+                        report.setOnAction(e -> HMCLService.openRedirectLink("multiplayer-feedback"));
                     });
             FXUtils.setLimitWidth(sideBar, 200);
             setLeft(sideBar);
@@ -256,7 +256,7 @@ public class MultiplayerPageSkin extends DecoratorAnimatedPage.DecoratorAnimated
                 tokenField.setPromptText(i18n("multiplayer.session.create.token.prompt"));
 
                 JFXHyperlink applyLink = new JFXHyperlink(i18n("multiplayer.session.create.token.apply"));
-                applyLink.setOnAction(e -> FXUtils.openLink("https://noin.cn/circle/386.html"));
+                applyLink.setOnAction(e -> HMCLService.openRedirectLink("multiplayer-static-token"));
 
                 gridPane.addRow(0, new Label(i18n("multiplayer.session.create.token")), tokenField, applyLink);
 
@@ -270,7 +270,7 @@ public class MultiplayerPageSkin extends DecoratorAnimatedPage.DecoratorAnimated
                 pane.setAlignment(Pos.CENTER_LEFT);
 
                 JFXHyperlink aboutLink = new JFXHyperlink(i18n("about"));
-                aboutLink.setOnAction(e -> FXUtils.openLink("https://noin.cn/71.html"));
+                aboutLink.setOnAction(e -> HMCLService.openRedirectLink("multiplayer-about"));
 
                 HBox placeholder = new HBox();
                 HBox.setHgrow(placeholder, Priority.ALWAYS);

HMCL/src/main/java/org/jackhuang/hmcl/util/HMCLService.java

@@ -0,0 +1,29 @@
+/*
+ * Hello Minecraft! Launcher
+ * Copyright (C) 2021  huangyuhui <huanghongxun2008@126.com> and contributors
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.jackhuang.hmcl.util;
+
+import org.jackhuang.hmcl.ui.FXUtils;
+
+public final class HMCLService {
+    private HMCLService() {
+    }
+
+    public static void openRedirectLink(String id) {
+        FXUtils.openLink("https://hmcl.huangyuhui.net/api/redirect/" + id);
+    }
+}

HMCL/src/main/resources/assets/lang/I18N_zh.properties

@@ -640,7 +640,7 @@ multiplayer.download.success=多人聯機初始化完成
 multiplayer.download.unsupported=多人聯機依賴不支持當前系統或平台
 multiplayer.exit.after_ready=多人聯機會話意外退出，退出碼 %d
 multiplayer.exit.before_ready=多人聯機房間創建失敗，cato 退出碼 %d
-multiplayer.exit.timeout=無法連接多人聯機服務
+multiplayer.exit.timeout=無法連接多人聯機服務，你可以在多人聯機頁面的回饋中回饋問題。
 multiplayer.hint=多人聯機功能處於實驗階段，如果有問題請回饋。
 multiplayer.nat=網路檢測
 multiplayer.nat.hint=執行網路檢測可以讓你更清楚你的網路狀況是否符合聯機功能的需求。不符合聯機功能運行條件的網路狀況將可能導致聯機失敗。

HMCL/src/main/resources/assets/lang/I18N_zh_CN.properties

@@ -640,7 +640,7 @@ multiplayer.download.success=多人联机初始化完成
 multiplayer.download.unsupported=多人联机依赖不支持当前系统或平台
 multiplayer.exit.after_ready=多人联机会话意外退出，退出码 %d
 multiplayer.exit.before_ready=多人联机房间创建失败，cato 退出码 %d
-multiplayer.exit.timeout=无法连接多人联机服务
+multiplayer.exit.timeout=无法连接多人联机服务，你可以在多人联机页面的反馈中反馈问题。
 multiplayer.hint=多人联机功能处于实验阶段，如果有问题请反馈。
 multiplayer.nat=网络检测
 multiplayer.nat.hint=执行网络检测可以让你更清楚你的网络状况是否符合联机功能的需求。检测结果为差的网络可能导致联机失败。

HMCLCore/src/main/java/org/jackhuang/hmcl/event/EventManager.java

@@ -72,7 +72,7 @@ public final class EventManager<T extends Event> {
             return Event.Result.DEFAULT;
     }
 
-    private synchronized void removeConsumer(Consumer<T> consumer) {
+    public synchronized void unregister(Consumer<T> consumer) {
         handlers.removeValue(consumer);
     }
 
@@ -87,7 +87,7 @@ public final class EventManager<T extends Event> {
         public void accept(T t) {
             Consumer<T> listener = ref.get();
             if (listener == null) {
-                removeConsumer(this);
+                unregister(this);
             } else {
                 listener.accept(t);
             }