chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

typo, javadoc,code cleanup

Browse Source
This commit is contained in:
wener
2015-04-23 17:25:36 +08:00
parent d01e82ce3c
commit bdf3b0c861
13 changed files with 57 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
common/src/main/java/com/alibaba/otter/canal/common/CanalLifeCycle.java
View File

@@ -6,9 +6,9 @@ package com.alibaba.otter.canal.common;
*/ */
public interface CanalLifeCycle { public interface CanalLifeCycle {
public void start(); void start();
public void stop(); void stop();
public boolean isStart(); boolean isStart();
} }

3
common/src/main/java/com/alibaba/otter/canal/common/alarm/CanalAlarmHandler.java
View File

@@ -14,8 +14,7 @@ public interface CanalAlarmHandler extends CanalLifeCycle {
* 发送对应destination的报警 * 发送对应destination的报警
* *
* @param destination * @param destination
* @param title
* @param msg * @param msg
*/ */
public void sendAlarm(String destination, String msg); void sendAlarm(String destination, String msg);
} }

2
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
View File

@@ -391,7 +391,7 @@ public class CanalController {
} }
if (autoScan) { if (autoScan) {
instanceConfigMonitors.get(config.getMode()).regeister(destination, defaultAction); instanceConfigMonitors.get(config.getMode()).register(destination, defaultAction);
} }
} }

6
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceAction.java
View File

@@ -11,15 +11,15 @@ public interface InstanceAction {
/** /**
* 启动destination * 启动destination
*/ */
public void start(String destination); void start(String destination);
/** /**
* 停止destination * 停止destination
*/ */
public void stop(String destination); void stop(String destination);
/** /**
* 重载destination，可能需要stop,start操作，或者只是更新下内存配置 * 重载destination，可能需要stop,start操作，或者只是更新下内存配置
*/ */
public void reload(String destination); void reload(String destination);
} }

4
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceConfigMonitor.java
View File

@@ -10,7 +10,7 @@ import com.alibaba.otter.canal.common.CanalLifeCycle;
*/ */
public interface InstanceConfigMonitor extends 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);
} }

4
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java
View File

@@ -9,11 +9,11 @@ import com.alibaba.otter.canal.common.CanalLifeCycle;
*/ */
public class ManagerInstanceConfigMonitor extends AbstractCanalLifeCycle implements InstanceConfigMonitor, 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) {
} }

4
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java
View File

@@ -72,7 +72,7 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
lastFiles.clear(); lastFiles.clear();
} }
public void regeister(String destination, InstanceAction action) { public void register(String destination, InstanceAction action) {
if (action != null) { if (action != null) {
actions.put(destination, action); actions.put(destination, action);
} else { } else {
@@ -80,7 +80,7 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
} }
} }
public void unRegeister(String destination) { public void unregister(String destination) {
actions.remove(destination); actions.remove(destination);
} }

14
instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalInstance.java
View File

@@ -16,20 +16,20 @@ import com.alibaba.otter.canal.store.CanalEventStore;
*/ */
public interface CanalInstance extends CanalLifeCycle { 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);
} }

4
server/src/main/java/com/alibaba/otter/canal/server/CanalServer.java
View File

@@ -11,7 +11,7 @@ import com.alibaba.otter.canal.server.exception.CanalServerException;
*/ */
public interface CanalServer extends CanalLifeCycle { public interface CanalServer extends CanalLifeCycle {
public void start() throws CanalServerException; void start() throws CanalServerException;
public void stop() throws CanalServerException; void stop() throws CanalServerException;
} }

9
server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java
View File

@@ -50,8 +50,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
canalInstances = new MapMaker().makeComputingMap(new Function<String, CanalInstance>() { canalInstances = new MapMaker().makeComputingMap(new Function<String, CanalInstance>() {
public CanalInstance apply(String destination) { public CanalInstance apply(String destination) {
CanalInstance canalInstance = canalInstanceGenerator.generate(destination); return canalInstanceGenerator.generate(destination);
return canalInstance;
} }
}); });
@@ -74,7 +73,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
} }
} }
} catch (Exception e) { } 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 { public void rollback(ClientIdentity clientIdentity) throws CanalServerException {
checkStart(clientIdentity.getDestination()); 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 { public void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException {
checkStart(clientIdentity.getDestination()); checkStart(clientIdentity.getDestination());

14
server/src/main/java/com/alibaba/otter/canal/server/netty/handler/ClientAuthenticationHandler.java
View File

@@ -33,14 +33,14 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler {
private static final Logger logger = LoggerFactory.getLogger(ClientAuthenticationHandler.class); private static final Logger logger = LoggerFactory.getLogger(ClientAuthenticationHandler.class);
private final int SUPPORTED_VERSION = 3; private final int SUPPORTED_VERSION = 3;
private final int defaultSubscriptorDisconnectIdleTimeout = 5 * 60 * 1000; private final int defaultSubscriptorDisconnectIdleTimeout = 5 * 60 * 1000;
private CanalServerWithEmbedded embededServer; private CanalServerWithEmbedded embeddedServer;
public ClientAuthenticationHandler(){ public ClientAuthenticationHandler(){
} }
public ClientAuthenticationHandler(CanalServerWithEmbedded embededServer){ public ClientAuthenticationHandler(CanalServerWithEmbedded embeddedServer){
this.embededServer = embededServer; this.embeddedServer = embeddedServer;
} }
public void messageReceived(final ChannelHandlerContext ctx, MessageEvent e) throws Exception { public void messageReceived(final ChannelHandlerContext ctx, MessageEvent e) throws Exception {
@@ -58,10 +58,10 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler {
clientAuth.getFilter()); clientAuth.getFilter());
try { try {
MDC.put("destination", clientIdentity.getDestination()); MDC.put("destination", clientIdentity.getDestination());
embededServer.subscribe(clientIdentity); embeddedServer.subscribe(clientIdentity);
ctx.setAttachment(clientIdentity);// 设置状态数据 ctx.setAttachment(clientIdentity);// 设置状态数据
// 尝试启动，如果已经启动，忽略 // 尝试启动，如果已经启动，忽略
if (!embededServer.isStart(clientIdentity.getDestination())) { if (!embeddedServer.isStart(clientIdentity.getDestination())) {
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity
.getDestination()); .getDestination());
if (!runningMonitor.isStart()) { if (!runningMonitor.isStart()) {
@@ -115,8 +115,8 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler {
} }
} }
public void setEmbededServer(CanalServerWithEmbedded embededServer) { public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
this.embededServer = embededServer; this.embeddedServer = embeddedServer;
} }
} }

36
server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java
View File

@@ -43,14 +43,12 @@ import com.alibaba.otter.canal.server.netty.NettyUtils;
public class SessionHandler extends SimpleChannelHandler { public class SessionHandler extends SimpleChannelHandler {
private static final Logger logger = LoggerFactory.getLogger(SessionHandler.class); private static final Logger logger = LoggerFactory.getLogger(SessionHandler.class);
private CanalServerWithEmbedded embededServer; private CanalServerWithEmbedded embeddedServer;
public SessionHandler(){ public SessionHandler(){ }
} public SessionHandler(CanalServerWithEmbedded embeddedServer){
this.embeddedServer = embeddedServer;
public SessionHandler(CanalServerWithEmbedded embededServer){
this.embededServer = embededServer;
} }
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { 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()), clientIdentity = new ClientIdentity(sub.getDestination(), Short.valueOf(sub.getClientId()),
sub.getFilter()); sub.getFilter());
MDC.put("destination", clientIdentity.getDestination()); 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()); ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination());
if (!runningMonitor.isStart()) { if (!runningMonitor.isStart()) {
runningMonitor.start(); runningMonitor.start();
@@ -90,7 +88,7 @@ public class SessionHandler extends SimpleChannelHandler {
clientIdentity = new ClientIdentity(unsub.getDestination(), Short.valueOf(unsub.getClientId()), clientIdentity = new ClientIdentity(unsub.getDestination(), Short.valueOf(unsub.getClientId()),
unsub.getFilter()); unsub.getFilter());
MDC.put("destination", clientIdentity.getDestination()); MDC.put("destination", clientIdentity.getDestination());
embededServer.unsubscribe(clientIdentity); embeddedServer.unsubscribe(clientIdentity);
stopCanalInstanceIfNecessary(clientIdentity);// 尝试关闭 stopCanalInstanceIfNecessary(clientIdentity);// 尝试关闭
NettyUtils.ack(ctx.getChannel(), null); NettyUtils.ack(ctx.getChannel(), null);
} else { } else {
@@ -108,17 +106,17 @@ public class SessionHandler extends SimpleChannelHandler {
// if (get.getAutoAck()) { // if (get.getAutoAck()) {
// if (get.getTimeout() == -1) {//是否是初始值 // if (get.getTimeout() == -1) {//是否是初始值
// message = embededServer.get(clientIdentity, get.getFetchSize()); // message = embeddedServer.get(clientIdentity, get.getFetchSize());
// } else { // } else {
// TimeUnit unit = convertTimeUnit(get.getUnit()); // TimeUnit unit = convertTimeUnit(get.getUnit());
// message = embededServer.get(clientIdentity, get.getFetchSize(), get.getTimeout(), unit); // message = embeddedServer.get(clientIdentity, get.getFetchSize(), get.getTimeout(), unit);
// } // }
// } else { // } else {
if (get.getTimeout() == -1) {//是否是初始值 if (get.getTimeout() == -1) {//是否是初始值
message = embededServer.getWithoutAck(clientIdentity, get.getFetchSize()); message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize());
} else { } else {
TimeUnit unit = convertTimeUnit(get.getUnit()); TimeUnit unit = convertTimeUnit(get.getUnit());
message = embededServer.getWithoutAck(clientIdentity, get.getFetchSize(), get.getTimeout(), message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize(), get.getTimeout(),
unit); unit);
} }
// } // }
@@ -153,7 +151,7 @@ public class SessionHandler extends SimpleChannelHandler {
// donothing // donothing
} else { } else {
clientIdentity = new ClientIdentity(ack.getDestination(), Short.valueOf(ack.getClientId())); clientIdentity = new ClientIdentity(ack.getDestination(), Short.valueOf(ack.getClientId()));
embededServer.ack(clientIdentity, ack.getBatchId()); embeddedServer.ack(clientIdentity, ack.getBatchId());
} }
} else { } else {
NettyUtils.error(401, NettyUtils.error(401,
@@ -169,9 +167,9 @@ public class SessionHandler extends SimpleChannelHandler {
clientIdentity = new ClientIdentity(rollback.getDestination(), clientIdentity = new ClientIdentity(rollback.getDestination(),
Short.valueOf(rollback.getClientId())); Short.valueOf(rollback.getClientId()));
if (rollback.getBatchId() == 0L) { if (rollback.getBatchId() == 0L) {
embededServer.rollback(clientIdentity);// 回滚所有批次 embeddedServer.rollback(clientIdentity);// 回滚所有批次
} else { } else {
embededServer.rollback(clientIdentity, rollback.getBatchId()); // 只回滚单个批次 embeddedServer.rollback(clientIdentity, rollback.getBatchId()); // 只回滚单个批次
} }
} else { } else {
NettyUtils.error(401, NettyUtils.error(401,
@@ -212,7 +210,7 @@ public class SessionHandler extends SimpleChannelHandler {
} }
private void stopCanalInstanceIfNecessary(ClientIdentity clientIdentity) { private void stopCanalInstanceIfNecessary(ClientIdentity clientIdentity) {
List<ClientIdentity> clientIdentitys = embededServer.listAllSubscribe(clientIdentity.getDestination()); List<ClientIdentity> clientIdentitys = embeddedServer.listAllSubscribe(clientIdentity.getDestination());
if (clientIdentitys != null && clientIdentitys.size() == 1 && clientIdentitys.contains(clientIdentity)) { if (clientIdentitys != null && clientIdentitys.size() == 1 && clientIdentitys.contains(clientIdentity)) {
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination()); ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination());
if (runningMonitor.isStart()) { if (runningMonitor.isStart()) {
@@ -242,8 +240,8 @@ public class SessionHandler extends SimpleChannelHandler {
} }
} }
public void setEmbededServer(CanalServerWithEmbedded embededServer) { public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
this.embededServer = embededServer; this.embeddedServer = embeddedServer;
} }
} }

10
server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithNettyTest.java
View File

@@ -49,16 +49,18 @@ public class CanalServerWithNettyTest {
@Before @Before
public void setUp() { public void setUp() {
CanalServerWithEmbedded embededServer = new CanalServerWithEmbedded(); CanalServerWithEmbedded embeddedServer = new CanalServerWithEmbedded();
embededServer.setCanalInstanceGenerator(new CanalInstanceGenerator() { embeddedServer.setCanalInstanceGenerator(new CanalInstanceGenerator()
{
public CanalInstance generate(String destination) { public CanalInstance generate(String destination)
{
Canal canal = buildCanal(); Canal canal = buildCanal();
return new CanalInstanceWithManager(canal, FILTER); return new CanalInstanceWithManager(canal, FILTER);
} }
}); });
nettyServer = new CanalServerWithNetty(embededServer); nettyServer = new CanalServerWithNetty(embeddedServer);
nettyServer.setPort(1088); nettyServer.setPort(1088);
nettyServer.start(); nettyServer.start();
} }