From bdf3b0c86164dd042e4d3f3adf05f2dd3cab84d2 Mon Sep 17 00:00:00 2001 From: wener Date: Thu, 23 Apr 2015 17:25:36 +0800 Subject: [PATCH] typo, javadoc,code cleanup --- .../otter/canal/common/CanalLifeCycle.java | 6 ++-- .../canal/common/alarm/CanalAlarmHandler.java | 3 +- .../otter/canal/deployer/CanalController.java | 2 +- .../deployer/monitor/InstanceAction.java | 6 ++-- .../monitor/InstanceConfigMonitor.java | 4 +-- .../monitor/ManagerInstanceConfigMonitor.java | 4 +-- .../monitor/SpringInstanceConfigMonitor.java | 4 +-- .../canal/instance/core/CanalInstance.java | 14 ++++---- .../otter/canal/server/CanalServer.java | 4 +-- .../embedded/CanalServerWithEmbedded.java | 9 +++-- .../handler/ClientAuthenticationHandler.java | 14 ++++---- .../server/netty/handler/SessionHandler.java | 36 +++++++++---------- .../server/CanalServerWithNettyTest.java | 10 +++--- 13 files changed, 57 insertions(+), 59 deletions(-) diff --git a/common/src/main/java/com/alibaba/otter/canal/common/CanalLifeCycle.java b/common/src/main/java/com/alibaba/otter/canal/common/CanalLifeCycle.java index 6bd15f14..da62d885 100644 --- a/common/src/main/java/com/alibaba/otter/canal/common/CanalLifeCycle.java +++ b/common/src/main/java/com/alibaba/otter/canal/common/CanalLifeCycle.java @@ -6,9 +6,9 @@ package com.alibaba.otter.canal.common; */ public interface CanalLifeCycle { - public void start(); + void start(); - public void stop(); + void stop(); - public boolean isStart(); + boolean isStart(); } diff --git a/common/src/main/java/com/alibaba/otter/canal/common/alarm/CanalAlarmHandler.java b/common/src/main/java/com/alibaba/otter/canal/common/alarm/CanalAlarmHandler.java index 4c9d1099..67e47f0c 100644 --- a/common/src/main/java/com/alibaba/otter/canal/common/alarm/CanalAlarmHandler.java +++ b/common/src/main/java/com/alibaba/otter/canal/common/alarm/CanalAlarmHandler.java @@ -14,8 +14,7 @@ public interface CanalAlarmHandler extends CanalLifeCycle { * 发送对应destination的报警 * * @param destination - * @param title * @param msg */ - public void sendAlarm(String destination, String msg); + void sendAlarm(String destination, String msg); } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java index 9246e26d..3e8f0c8a 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java @@ -391,7 +391,7 @@ public class CanalController { } if (autoScan) { - instanceConfigMonitors.get(config.getMode()).regeister(destination, defaultAction); + instanceConfigMonitors.get(config.getMode()).register(destination, defaultAction); } } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceAction.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceAction.java index 14551ea7..6f1112dc 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceAction.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceAction.java @@ -11,15 +11,15 @@ public interface InstanceAction { /** * 启动destination */ - public void start(String destination); + void start(String destination); /** * 停止destination */ - public void stop(String destination); + void stop(String destination); /** * 重载destination，可能需要stop,start操作，或者只是更新下内存配置 */ - public void reload(String destination); + void reload(String destination); } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceConfigMonitor.java index c1ec67d1..190c804c 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceConfigMonitor.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceConfigMonitor.java @@ -10,7 +10,7 @@ import com.alibaba.otter.canal.common.CanalLifeCycle; */ public interface InstanceConfigMonitor extends CanalLifeCycle { - public void regeister(String destination, InstanceAction action); + void register(String destination, InstanceAction action); - public void unRegeister(String destination); + void unregister(String destination); } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java index ee508b9f..d6099f29 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java @@ -9,11 +9,11 @@ import com.alibaba.otter.canal.common.CanalLifeCycle; */ public class ManagerInstanceConfigMonitor extends AbstractCanalLifeCycle implements InstanceConfigMonitor, CanalLifeCycle { - public void regeister(String destination, InstanceAction action) { + public void register(String destination, InstanceAction action) { } - public void unRegeister(String destination) { + public void unregister(String destination) { } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java index e19323a2..c56edf21 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java @@ -72,7 +72,7 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen lastFiles.clear(); } - public void regeister(String destination, InstanceAction action) { + public void register(String destination, InstanceAction action) { if (action != null) { actions.put(destination, action); } else { @@ -80,7 +80,7 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen } } - public void unRegeister(String destination) { + public void unregister(String destination) { actions.remove(destination); } diff --git a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalInstance.java b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalInstance.java index 198071a0..3bd0464b 100644 --- a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalInstance.java +++ b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalInstance.java @@ -16,20 +16,20 @@ import com.alibaba.otter.canal.store.CanalEventStore; */ public interface CanalInstance extends CanalLifeCycle { - public String getDestination(); + String getDestination(); - public CanalEventParser getEventParser(); + CanalEventParser getEventParser(); - public CanalEventSink getEventSink(); + CanalEventSink getEventSink(); - public CanalEventStore getEventStore(); + CanalEventStore getEventStore(); - public CanalMetaManager getMetaManager(); + CanalMetaManager getMetaManager(); - public CanalAlarmHandler getAlarmHandler(); + CanalAlarmHandler getAlarmHandler(); /** * 客户端发生订阅/取消订阅行为 */ - public boolean subscribeChange(ClientIdentity identity); + boolean subscribeChange(ClientIdentity identity); } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/CanalServer.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalServer.java index a6453d7b..ad067500 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/CanalServer.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalServer.java @@ -11,7 +11,7 @@ import com.alibaba.otter.canal.server.exception.CanalServerException; */ public interface CanalServer extends CanalLifeCycle { - public void start() throws CanalServerException; + void start() throws CanalServerException; - public void stop() throws CanalServerException; + void stop() throws CanalServerException; } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java index 30e827e9..604a54b6 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java @@ -50,8 +50,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C canalInstances = new MapMaker().makeComputingMap(new Function() { public CanalInstance apply(String destination) { - CanalInstance canalInstance = canalInstanceGenerator.generate(destination); - return canalInstance; + return canalInstanceGenerator.generate(destination); } }); @@ -74,7 +73,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C } } } catch (Exception e) { - logger.error(String.format("stop cannalInstance[%s] has an error", entry.getKey()), e); + logger.error(String.format("stop CanalInstance[%s] has an error", entry.getKey()), e); } } } @@ -357,7 +356,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C } /** - * 回滚到未进行 {@link ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link ack} 的地方开始拿 + * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿 */ public void rollback(ClientIdentity clientIdentity) throws CanalServerException { checkStart(clientIdentity.getDestination()); @@ -378,7 +377,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C } /** - * 回滚到未进行 {@link ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link ack} 的地方开始拿 + * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿 */ public void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException { checkStart(clientIdentity.getDestination()); diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/ClientAuthenticationHandler.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/ClientAuthenticationHandler.java index a925e4a6..9d86eb21 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/ClientAuthenticationHandler.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/ClientAuthenticationHandler.java @@ -33,14 +33,14 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler { private static final Logger logger = LoggerFactory.getLogger(ClientAuthenticationHandler.class); private final int SUPPORTED_VERSION = 3; private final int defaultSubscriptorDisconnectIdleTimeout = 5 * 60 * 1000; - private CanalServerWithEmbedded embededServer; + private CanalServerWithEmbedded embeddedServer; public ClientAuthenticationHandler(){ } - public ClientAuthenticationHandler(CanalServerWithEmbedded embededServer){ - this.embededServer = embededServer; + public ClientAuthenticationHandler(CanalServerWithEmbedded embeddedServer){ + this.embeddedServer = embeddedServer; } public void messageReceived(final ChannelHandlerContext ctx, MessageEvent e) throws Exception { @@ -58,10 +58,10 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler { clientAuth.getFilter()); try { MDC.put("destination", clientIdentity.getDestination()); - embededServer.subscribe(clientIdentity); + embeddedServer.subscribe(clientIdentity); ctx.setAttachment(clientIdentity);// 设置状态数据 // 尝试启动，如果已经启动，忽略 - if (!embededServer.isStart(clientIdentity.getDestination())) { + if (!embeddedServer.isStart(clientIdentity.getDestination())) { ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity .getDestination()); if (!runningMonitor.isStart()) { @@ -115,8 +115,8 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler { } } - public void setEmbededServer(CanalServerWithEmbedded embededServer) { - this.embededServer = embededServer; + public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) { + this.embeddedServer = embeddedServer; } } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java index bf83c6ff..cda44668 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java @@ -43,14 +43,12 @@ import com.alibaba.otter.canal.server.netty.NettyUtils; public class SessionHandler extends SimpleChannelHandler { private static final Logger logger = LoggerFactory.getLogger(SessionHandler.class); - private CanalServerWithEmbedded embededServer; + private CanalServerWithEmbedded embeddedServer; - public SessionHandler(){ + public SessionHandler(){ } - } - - public SessionHandler(CanalServerWithEmbedded embededServer){ - this.embededServer = embededServer; + public SessionHandler(CanalServerWithEmbedded embeddedServer){ + this.embeddedServer = embeddedServer; } public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { @@ -66,10 +64,10 @@ public class SessionHandler extends SimpleChannelHandler { clientIdentity = new ClientIdentity(sub.getDestination(), Short.valueOf(sub.getClientId()), sub.getFilter()); MDC.put("destination", clientIdentity.getDestination()); - embededServer.subscribe(clientIdentity); + embeddedServer.subscribe(clientIdentity); // 尝试启动，如果已经启动，忽略 - if (!embededServer.isStart(clientIdentity.getDestination())) { + if (!embeddedServer.isStart(clientIdentity.getDestination())) { ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination()); if (!runningMonitor.isStart()) { runningMonitor.start(); @@ -90,7 +88,7 @@ public class SessionHandler extends SimpleChannelHandler { clientIdentity = new ClientIdentity(unsub.getDestination(), Short.valueOf(unsub.getClientId()), unsub.getFilter()); MDC.put("destination", clientIdentity.getDestination()); - embededServer.unsubscribe(clientIdentity); + embeddedServer.unsubscribe(clientIdentity); stopCanalInstanceIfNecessary(clientIdentity);// 尝试关闭 NettyUtils.ack(ctx.getChannel(), null); } else { @@ -108,17 +106,17 @@ public class SessionHandler extends SimpleChannelHandler { // if (get.getAutoAck()) { // if (get.getTimeout() == -1) {//是否是初始值 - // message = embededServer.get(clientIdentity, get.getFetchSize()); + // message = embeddedServer.get(clientIdentity, get.getFetchSize()); // } else { // TimeUnit unit = convertTimeUnit(get.getUnit()); - // message = embededServer.get(clientIdentity, get.getFetchSize(), get.getTimeout(), unit); + // message = embeddedServer.get(clientIdentity, get.getFetchSize(), get.getTimeout(), unit); // } // } else { if (get.getTimeout() == -1) {//是否是初始值 - message = embededServer.getWithoutAck(clientIdentity, get.getFetchSize()); + message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize()); } else { TimeUnit unit = convertTimeUnit(get.getUnit()); - message = embededServer.getWithoutAck(clientIdentity, get.getFetchSize(), get.getTimeout(), + message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize(), get.getTimeout(), unit); } // } @@ -153,7 +151,7 @@ public class SessionHandler extends SimpleChannelHandler { // donothing } else { clientIdentity = new ClientIdentity(ack.getDestination(), Short.valueOf(ack.getClientId())); - embededServer.ack(clientIdentity, ack.getBatchId()); + embeddedServer.ack(clientIdentity, ack.getBatchId()); } } else { NettyUtils.error(401, @@ -169,9 +167,9 @@ public class SessionHandler extends SimpleChannelHandler { clientIdentity = new ClientIdentity(rollback.getDestination(), Short.valueOf(rollback.getClientId())); if (rollback.getBatchId() == 0L) { - embededServer.rollback(clientIdentity);// 回滚所有批次 + embeddedServer.rollback(clientIdentity);// 回滚所有批次 } else { - embededServer.rollback(clientIdentity, rollback.getBatchId()); // 只回滚单个批次 + embeddedServer.rollback(clientIdentity, rollback.getBatchId()); // 只回滚单个批次 } } else { NettyUtils.error(401, @@ -212,7 +210,7 @@ public class SessionHandler extends SimpleChannelHandler { } private void stopCanalInstanceIfNecessary(ClientIdentity clientIdentity) { - List clientIdentitys = embededServer.listAllSubscribe(clientIdentity.getDestination()); + List clientIdentitys = embeddedServer.listAllSubscribe(clientIdentity.getDestination()); if (clientIdentitys != null && clientIdentitys.size() == 1 && clientIdentitys.contains(clientIdentity)) { ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination()); if (runningMonitor.isStart()) { @@ -242,8 +240,8 @@ public class SessionHandler extends SimpleChannelHandler { } } - public void setEmbededServer(CanalServerWithEmbedded embededServer) { - this.embededServer = embededServer; + public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) { + this.embeddedServer = embeddedServer; } } diff --git a/server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithNettyTest.java b/server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithNettyTest.java index 102436c5..78032adb 100644 --- a/server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithNettyTest.java +++ b/server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithNettyTest.java @@ -49,16 +49,18 @@ public class CanalServerWithNettyTest { @Before public void setUp() { - CanalServerWithEmbedded embededServer = new CanalServerWithEmbedded(); - embededServer.setCanalInstanceGenerator(new CanalInstanceGenerator() { + CanalServerWithEmbedded embeddedServer = new CanalServerWithEmbedded(); + embeddedServer.setCanalInstanceGenerator(new CanalInstanceGenerator() + { - public CanalInstance generate(String destination) { + public CanalInstance generate(String destination) + { Canal canal = buildCanal(); return new CanalInstanceWithManager(canal, FILTER); } }); - nettyServer = new CanalServerWithNetty(embededServer); + nettyServer = new CanalServerWithNetty(embeddedServer); nettyServer.setPort(1088); nettyServer.start(); }