mirror of
https://github.com/alibaba/canal.git
synced 2026-11-04 00:30:15 +08:00
fix typo embeded -> embedded
This commit is contained in:
@@ -33,7 +33,7 @@ import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator;
|
|||||||
import com.alibaba.otter.canal.instance.manager.CanalConfigClient;
|
import com.alibaba.otter.canal.instance.manager.CanalConfigClient;
|
||||||
import com.alibaba.otter.canal.instance.manager.ManagerCanalInstanceGenerator;
|
import com.alibaba.otter.canal.instance.manager.ManagerCanalInstanceGenerator;
|
||||||
import com.alibaba.otter.canal.instance.spring.SpringCanalInstanceGenerator;
|
import com.alibaba.otter.canal.instance.spring.SpringCanalInstanceGenerator;
|
||||||
import com.alibaba.otter.canal.server.embeded.CanalServerWithEmbeded;
|
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
|
||||||
import com.alibaba.otter.canal.server.exception.CanalServerException;
|
import com.alibaba.otter.canal.server.exception.CanalServerException;
|
||||||
import com.alibaba.otter.canal.server.netty.CanalServerWithNetty;
|
import com.alibaba.otter.canal.server.netty.CanalServerWithNetty;
|
||||||
import com.google.common.base.Function;
|
import com.google.common.base.Function;
|
||||||
@@ -59,7 +59,7 @@ public class CanalController {
|
|||||||
private boolean autoScan = true;
|
private boolean autoScan = true;
|
||||||
private InstanceAction defaultAction;
|
private InstanceAction defaultAction;
|
||||||
private Map<InstanceMode, InstanceConfigMonitor> instanceConfigMonitors;
|
private Map<InstanceMode, InstanceConfigMonitor> instanceConfigMonitors;
|
||||||
private CanalServerWithEmbeded embededCanalServer;
|
private CanalServerWithEmbedded embededCanalServer;
|
||||||
private CanalServerWithNetty canalServer;
|
private CanalServerWithNetty canalServer;
|
||||||
|
|
||||||
private CanalInstanceGenerator instanceGenerator;
|
private CanalInstanceGenerator instanceGenerator;
|
||||||
@@ -87,7 +87,7 @@ public class CanalController {
|
|||||||
cid = Long.valueOf(getProperty(properties, CanalConstants.CANAL_ID));
|
cid = Long.valueOf(getProperty(properties, CanalConstants.CANAL_ID));
|
||||||
ip = getProperty(properties, CanalConstants.CANAL_IP);
|
ip = getProperty(properties, CanalConstants.CANAL_IP);
|
||||||
port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT));
|
port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT));
|
||||||
embededCanalServer = new CanalServerWithEmbeded();
|
embededCanalServer = new CanalServerWithEmbedded();
|
||||||
embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator
|
embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator
|
||||||
canalServer = new CanalServerWithNetty(embededCanalServer);
|
canalServer = new CanalServerWithNetty(embededCanalServer);
|
||||||
canalServer.setIp(ip);
|
canalServer.setIp(ip);
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
package com.alibaba.otter.canal.server.embeded;
|
package com.alibaba.otter.canal.server.embedded;
|
||||||
|
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.Collections;
|
import java.util.Collections;
|
||||||
@@ -37,9 +37,9 @@ import com.google.common.collect.Maps;
|
|||||||
* @author zebin.xuzb
|
* @author zebin.xuzb
|
||||||
* @version 1.0.0
|
* @version 1.0.0
|
||||||
*/
|
*/
|
||||||
public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements CanalServer {
|
public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer {
|
||||||
|
|
||||||
private static final Logger logger = LoggerFactory.getLogger(CanalServerWithEmbeded.class);
|
private static final Logger logger = LoggerFactory.getLogger(CanalServerWithEmbedded.class);
|
||||||
private Map<String, CanalInstance> canalInstances;
|
private Map<String, CanalInstance> canalInstances;
|
||||||
// private Map<ClientIdentity, Position> lastRollbackPostions;
|
// private Map<ClientIdentity, Position> lastRollbackPostions;
|
||||||
private CanalInstanceGenerator canalInstanceGenerator;
|
private CanalInstanceGenerator canalInstanceGenerator;
|
||||||
@@ -13,7 +13,7 @@ import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
|
|||||||
|
|
||||||
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
|
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
|
||||||
import com.alibaba.otter.canal.server.CanalServer;
|
import com.alibaba.otter.canal.server.CanalServer;
|
||||||
import com.alibaba.otter.canal.server.embeded.CanalServerWithEmbeded;
|
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
|
||||||
import com.alibaba.otter.canal.server.netty.handler.ClientAuthenticationHandler;
|
import com.alibaba.otter.canal.server.netty.handler.ClientAuthenticationHandler;
|
||||||
import com.alibaba.otter.canal.server.netty.handler.FixedHeaderFrameDecoder;
|
import com.alibaba.otter.canal.server.netty.handler.FixedHeaderFrameDecoder;
|
||||||
import com.alibaba.otter.canal.server.netty.handler.HandshakeInitializationHandler;
|
import com.alibaba.otter.canal.server.netty.handler.HandshakeInitializationHandler;
|
||||||
@@ -27,7 +27,7 @@ import com.alibaba.otter.canal.server.netty.handler.SessionHandler;
|
|||||||
*/
|
*/
|
||||||
public class CanalServerWithNetty extends AbstractCanalLifeCycle implements CanalServer {
|
public class CanalServerWithNetty extends AbstractCanalLifeCycle implements CanalServer {
|
||||||
|
|
||||||
private CanalServerWithEmbeded embededServer; // 嵌入式server
|
private CanalServerWithEmbedded embeddedServer; // 嵌入式server
|
||||||
private String ip;
|
private String ip;
|
||||||
private int port;
|
private int port;
|
||||||
private Channel serverChannel = null;
|
private Channel serverChannel = null;
|
||||||
@@ -36,15 +36,15 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
|
|||||||
public CanalServerWithNetty(){
|
public CanalServerWithNetty(){
|
||||||
}
|
}
|
||||||
|
|
||||||
public CanalServerWithNetty(CanalServerWithEmbeded embededServer){
|
public CanalServerWithNetty(CanalServerWithEmbedded embeddedServer){
|
||||||
this.embededServer = embededServer;
|
this.embeddedServer = embeddedServer;
|
||||||
}
|
}
|
||||||
|
|
||||||
public void start() {
|
public void start() {
|
||||||
super.start();
|
super.start();
|
||||||
|
|
||||||
if (!embededServer.isStart()) {
|
if (!embeddedServer.isStart()) {
|
||||||
embededServer.start();
|
embeddedServer.start();
|
||||||
}
|
}
|
||||||
|
|
||||||
this.bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
|
this.bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
|
||||||
@@ -58,9 +58,9 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
|
|||||||
pipelines.addLast(FixedHeaderFrameDecoder.class.getName(), new FixedHeaderFrameDecoder());
|
pipelines.addLast(FixedHeaderFrameDecoder.class.getName(), new FixedHeaderFrameDecoder());
|
||||||
pipelines.addLast(HandshakeInitializationHandler.class.getName(), new HandshakeInitializationHandler());
|
pipelines.addLast(HandshakeInitializationHandler.class.getName(), new HandshakeInitializationHandler());
|
||||||
pipelines.addLast(ClientAuthenticationHandler.class.getName(),
|
pipelines.addLast(ClientAuthenticationHandler.class.getName(),
|
||||||
new ClientAuthenticationHandler(embededServer));
|
new ClientAuthenticationHandler(embeddedServer));
|
||||||
|
|
||||||
SessionHandler sessionHandler = new SessionHandler(embededServer);
|
SessionHandler sessionHandler = new SessionHandler(embeddedServer);
|
||||||
pipelines.addLast(SessionHandler.class.getName(), sessionHandler);
|
pipelines.addLast(SessionHandler.class.getName(), sessionHandler);
|
||||||
return pipelines;
|
return pipelines;
|
||||||
}
|
}
|
||||||
@@ -85,8 +85,8 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
|
|||||||
this.bootstrap.releaseExternalResources();
|
this.bootstrap.releaseExternalResources();
|
||||||
}
|
}
|
||||||
|
|
||||||
if (embededServer.isStart()) {
|
if (embeddedServer.isStart()) {
|
||||||
embededServer.stop();
|
embeddedServer.stop();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -98,8 +98,8 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
|
|||||||
this.port = port;
|
this.port = port;
|
||||||
}
|
}
|
||||||
|
|
||||||
public void setEmbededServer(CanalServerWithEmbeded embededServer) {
|
public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
|
||||||
this.embededServer = embededServer;
|
this.embeddedServer = embeddedServer;
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,7 +19,7 @@ import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitors;
|
|||||||
import com.alibaba.otter.canal.protocol.CanalPacket.ClientAuth;
|
import com.alibaba.otter.canal.protocol.CanalPacket.ClientAuth;
|
||||||
import com.alibaba.otter.canal.protocol.CanalPacket.Packet;
|
import com.alibaba.otter.canal.protocol.CanalPacket.Packet;
|
||||||
import com.alibaba.otter.canal.protocol.ClientIdentity;
|
import com.alibaba.otter.canal.protocol.ClientIdentity;
|
||||||
import com.alibaba.otter.canal.server.embeded.CanalServerWithEmbeded;
|
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
|
||||||
import com.alibaba.otter.canal.server.netty.NettyUtils;
|
import com.alibaba.otter.canal.server.netty.NettyUtils;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -33,13 +33,13 @@ 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 CanalServerWithEmbeded embededServer;
|
private CanalServerWithEmbedded embededServer;
|
||||||
|
|
||||||
public ClientAuthenticationHandler(){
|
public ClientAuthenticationHandler(){
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public ClientAuthenticationHandler(CanalServerWithEmbeded embededServer){
|
public ClientAuthenticationHandler(CanalServerWithEmbedded embededServer){
|
||||||
this.embededServer = embededServer;
|
this.embededServer = embededServer;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -62,7 +62,8 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler {
|
|||||||
ctx.setAttachment(clientIdentity);// 设置状态数据
|
ctx.setAttachment(clientIdentity);// 设置状态数据
|
||||||
// 尝试启动,如果已经启动,忽略
|
// 尝试启动,如果已经启动,忽略
|
||||||
if (!embededServer.isStart(clientIdentity.getDestination())) {
|
if (!embededServer.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();
|
||||||
}
|
}
|
||||||
@@ -114,7 +115,7 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
public void setEmbededServer(CanalServerWithEmbeded embededServer) {
|
public void setEmbededServer(CanalServerWithEmbedded embededServer) {
|
||||||
this.embededServer = embededServer;
|
this.embededServer = embededServer;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -31,7 +31,7 @@ import com.alibaba.otter.canal.protocol.CanalPacket.Sub;
|
|||||||
import com.alibaba.otter.canal.protocol.CanalPacket.Unsub;
|
import com.alibaba.otter.canal.protocol.CanalPacket.Unsub;
|
||||||
import com.alibaba.otter.canal.protocol.ClientIdentity;
|
import com.alibaba.otter.canal.protocol.ClientIdentity;
|
||||||
import com.alibaba.otter.canal.protocol.Message;
|
import com.alibaba.otter.canal.protocol.Message;
|
||||||
import com.alibaba.otter.canal.server.embeded.CanalServerWithEmbeded;
|
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
|
||||||
import com.alibaba.otter.canal.server.netty.NettyUtils;
|
import com.alibaba.otter.canal.server.netty.NettyUtils;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -43,13 +43,13 @@ 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 CanalServerWithEmbeded embededServer;
|
private CanalServerWithEmbedded embededServer;
|
||||||
|
|
||||||
public SessionHandler(){
|
public SessionHandler(){
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public SessionHandler(CanalServerWithEmbeded embededServer){
|
public SessionHandler(CanalServerWithEmbedded embededServer){
|
||||||
this.embededServer = embededServer;
|
this.embededServer = embededServer;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -242,7 +242,7 @@ public class SessionHandler extends SimpleChannelHandler {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
public void setEmbededServer(CanalServerWithEmbeded embededServer) {
|
public void setEmbededServer(CanalServerWithEmbedded embededServer) {
|
||||||
this.embededServer = embededServer;
|
this.embededServer = embededServer;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -14,7 +14,7 @@ import com.alibaba.otter.canal.parse.CanalEventParser;
|
|||||||
import com.alibaba.otter.canal.parse.CanalHASwitchable;
|
import com.alibaba.otter.canal.parse.CanalHASwitchable;
|
||||||
import com.alibaba.otter.canal.protocol.ClientIdentity;
|
import com.alibaba.otter.canal.protocol.ClientIdentity;
|
||||||
import com.alibaba.otter.canal.protocol.Message;
|
import com.alibaba.otter.canal.protocol.Message;
|
||||||
import com.alibaba.otter.canal.server.embeded.CanalServerWithEmbeded;
|
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
|
||||||
|
|
||||||
public abstract class BaseCanalServerWithEmbededTest {
|
public abstract class BaseCanalServerWithEmbededTest {
|
||||||
|
|
||||||
@@ -26,12 +26,12 @@ public abstract class BaseCanalServerWithEmbededTest {
|
|||||||
protected static final String PASSWORD = "retl";
|
protected static final String PASSWORD = "retl";
|
||||||
protected static final String FILTER = "retl\\..*,erosa.zk_complaint_bizdata";
|
protected static final String FILTER = "retl\\..*,erosa.zk_complaint_bizdata";
|
||||||
|
|
||||||
private CanalServerWithEmbeded server;
|
private CanalServerWithEmbedded server;
|
||||||
private ClientIdentity clientIdentity = new ClientIdentity(DESTINATION, (short) 1); ;
|
private ClientIdentity clientIdentity = new ClientIdentity(DESTINATION, (short) 1); ;
|
||||||
|
|
||||||
@Before
|
@Before
|
||||||
public void setUp() {
|
public void setUp() {
|
||||||
server = new CanalServerWithEmbeded();
|
server = new CanalServerWithEmbedded();
|
||||||
server.setCanalInstanceGenerator(new CanalInstanceGenerator() {
|
server.setCanalInstanceGenerator(new CanalInstanceGenerator() {
|
||||||
|
|
||||||
public CanalInstance generate(String destination) {
|
public CanalInstance generate(String destination) {
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ import com.alibaba.otter.canal.instance.manager.model.CanalParameter.MetaMode;
|
|||||||
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType;
|
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType;
|
||||||
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode;
|
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode;
|
||||||
|
|
||||||
public class CanalServerWithEmbeded_StandaloneTest extends BaseCanalServerWithEmbededTest {
|
public class CanalServerWithEmbedded_StandaloneTest extends BaseCanalServerWithEmbededTest {
|
||||||
|
|
||||||
protected Canal buildCanal() {
|
protected Canal buildCanal() {
|
||||||
Canal canal = new Canal();
|
Canal canal = new Canal();
|
||||||
@@ -15,7 +15,7 @@ import com.alibaba.otter.canal.instance.manager.model.CanalParameter.MetaMode;
|
|||||||
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType;
|
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType;
|
||||||
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode;
|
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode;
|
||||||
|
|
||||||
public class CanalServerWithEmbeded_StandbyTest extends BaseCanalServerWithEmbededTest {
|
public class CanalServerWithEmbedded_StandbyTest extends BaseCanalServerWithEmbededTest {
|
||||||
|
|
||||||
private ZkClient zkClient = new ZkClient(cluster1);
|
private ZkClient zkClient = new ZkClient(cluster1);
|
||||||
|
|
||||||
@@ -31,7 +31,7 @@ import com.alibaba.otter.canal.protocol.CanalPacket.Packet;
|
|||||||
import com.alibaba.otter.canal.protocol.CanalPacket.PacketType;
|
import com.alibaba.otter.canal.protocol.CanalPacket.PacketType;
|
||||||
import com.alibaba.otter.canal.protocol.CanalPacket.Sub;
|
import com.alibaba.otter.canal.protocol.CanalPacket.Sub;
|
||||||
import com.alibaba.otter.canal.protocol.CanalPacket.Unsub;
|
import com.alibaba.otter.canal.protocol.CanalPacket.Unsub;
|
||||||
import com.alibaba.otter.canal.server.embeded.CanalServerWithEmbeded;
|
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
|
||||||
import com.alibaba.otter.canal.server.netty.CanalServerWithNetty;
|
import com.alibaba.otter.canal.server.netty.CanalServerWithNetty;
|
||||||
|
|
||||||
public class CanalServerWithNettyTest {
|
public class CanalServerWithNettyTest {
|
||||||
@@ -49,7 +49,7 @@ public class CanalServerWithNettyTest {
|
|||||||
|
|
||||||
@Before
|
@Before
|
||||||
public void setUp() {
|
public void setUp() {
|
||||||
CanalServerWithEmbeded embededServer = new CanalServerWithEmbeded();
|
CanalServerWithEmbedded embededServer = new CanalServerWithEmbedded();
|
||||||
embededServer.setCanalInstanceGenerator(new CanalInstanceGenerator() {
|
embededServer.setCanalInstanceGenerator(new CanalInstanceGenerator() {
|
||||||
|
|
||||||
public CanalInstance generate(String destination) {
|
public CanalInstance generate(String destination) {
|
||||||
|
|||||||
Reference in New Issue
Block a user