diff --git a/src/CanalSharp/Connections/HA/CanalClientRunningInfo.cs b/src/CanalSharp/Connections/HA/CanalClientRunningInfo.cs
new file mode 100644
index 0000000..3f339fd
--- /dev/null
+++ b/src/CanalSharp/Connections/HA/CanalClientRunningInfo.cs
@@ -0,0 +1,16 @@
+using Newtonsoft.Json;
+
+namespace CanalSharp.Connections
+{
+ public class CanalClientRunningInfo
+ {
+ [JsonProperty("active")]
+ public bool Active { get; set; }
+
+ [JsonProperty("address")]
+ public string Address { get; set; }
+
+ [JsonProperty("clientId")]
+ public string ClientId { get; set; }
+ }
+}
\ No newline at end of file
diff --git a/src/CanalSharp/Connections/HA/CanalRunningInfo.cs b/src/CanalSharp/Connections/HA/CanalServerRunningInfo.cs
similarity index 77%
rename from src/CanalSharp/Connections/HA/CanalRunningInfo.cs
rename to src/CanalSharp/Connections/HA/CanalServerRunningInfo.cs
index 502cdfe..2e01e94 100644
--- a/src/CanalSharp/Connections/HA/CanalRunningInfo.cs
+++ b/src/CanalSharp/Connections/HA/CanalServerRunningInfo.cs
@@ -1,6 +1,6 @@
namespace CanalSharp.Connections
{
- public class CanalRunningInfo
+ public class CanalServerRunningInfo
{
public bool Active { get; set; }
public string Address { get; set; }
diff --git a/src/CanalSharp/Connections/HA/ClusterCanalConnection.cs b/src/CanalSharp/Connections/HA/ClusterCanalConnection.cs
index 5881d31..8053da1 100644
--- a/src/CanalSharp/Connections/HA/ClusterCanalConnection.cs
+++ b/src/CanalSharp/Connections/HA/ClusterCanalConnection.cs
@@ -1,4 +1,5 @@
using System;
+using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using CanalSharp.Connections.Enums;
@@ -7,36 +8,44 @@ using JetBrains.Annotations;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using NZookeeper;
+using NZookeeper.ACL;
using NZookeeper.Enums;
+using NZookeeper.Node;
+using org.apache.zookeeper;
namespace CanalSharp.Connections
{
///
/// Support canal server cluster and client cluster.
///
- public class ClusterCanalConnection : ICanalConnection
+ public class ClusterCanalConnection
{
private readonly ClusterCanalOptions _options;
private readonly ILoggerFactory _loggerFactory;
private readonly ILogger _logger;
+ private readonly TaskCompletionSource _completionSource;
// ReSharper disable once InconsistentNaming
- private readonly string ZK_RUNNING_NODE;
+ private readonly string ZK_SERVER_RUNNING_NODE;
+ // ReSharper disable once InconsistentNaming
+ private readonly string ZK_CLIENT_RUNNING_NODE;
- // For reconnection
+ // For reconnect
private string _lastSubFilter;
private bool _serverRunningNodeReCreated;
private SimpleCanalConnection _currentConn;
private ZkConnection _zk;
+ private CanalClientRunningInfo _clientRunningInfo;
public ClusterCanalConnection([NotNull] ClusterCanalOptions options, ILoggerFactory loggerFactory)
{
_options = options;
_loggerFactory = loggerFactory;
_logger = loggerFactory.CreateLogger();
- ZK_RUNNING_NODE = $"/otter/canal/destinations/{_options.Destination}/running";
+ _completionSource = new TaskCompletionSource();
+ ZK_SERVER_RUNNING_NODE = $"/otter/canal/destinations/{_options.Destination}/running";
+ ZK_CLIENT_RUNNING_NODE = $"/otter/canal/destinations/{_options.Destination}/{_options.ClientId}/running";
}
-
public async Task ReConnectAsync()
{
await DisConnectAsync();
@@ -72,17 +81,78 @@ namespace CanalSharp.Connections
public async Task ConnectAsync()
{
- await ConnectToZkAsync();
//get canal address from zk
- var nodeData = await _zk.GetDataAsync(ZK_RUNNING_NODE);
- var runningInfo = JsonConvert.DeserializeObject(Encoding.UTF8.GetString(nodeData));
+ await ConnectToZkAsync();
+ var nodeData = await _zk.GetDataAsync(ZK_SERVER_RUNNING_NODE);
+ var runningInfo = JsonConvert.DeserializeObject(Encoding.UTF8.GetString(nodeData));
_logger.LogInformation($"get canal address from zookeeper success: {runningInfo.Address}");
//connect to canal
_currentConn = new SimpleCanalConnection(CopyOptions(runningInfo), _loggerFactory.CreateLogger());
await _currentConn.ConnectAsync();
-
_serverRunningNodeReCreated = false;
+
+ var localIp = _currentConn.GetLocalEndPoint().ToString();
+ _clientRunningInfo = new CanalClientRunningInfo()
+ { Active = true, Address = localIp, ClientId = _options.ClientId };
+ _ = GetZkLockAsync(_clientRunningInfo);
+ await _completionSource.Task;
+
+ _logger.LogInformation("Ready to use!");
+ }
+
+ public async Task GetZkLockAsync(CanalClientRunningInfo runningInfo, bool waiting = false)
+ {
+ var times = 0;
+ while (waiting && times < 60)
+ {
+ await Task.Delay(1000);
+ times++;
+ _logger.LogWarning($"Waiting for get lock {times}-60...");
+ }
+
+ if (await _zk.NodeExistsAsync(ZK_CLIENT_RUNNING_NODE))
+ {
+ _logger.LogInformation($"Node {ZK_CLIENT_RUNNING_NODE} exits, get Zookeeper lock failed. Other instances are running.");
+ _logger.LogWarning("Waiting...");
+ }
+ else
+ {
+ try
+ {
+ var clientNodeData = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(runningInfo));
+
+ var parentNode = ZK_CLIENT_RUNNING_NODE.Replace("/running", "");
+ if (!await _zk.NodeExistsAsync(parentNode))
+ await _zk.CreateNodeAsync(parentNode, null,
+ new List() { new Acl(AclPerm.All, AclScheme.World, AclId.World()) }, NodeType.Persistent);
+
+ await _zk.CreateNodeAsync(ZK_CLIENT_RUNNING_NODE, clientNodeData,
+ new List() { new Acl(AclPerm.All, AclScheme.World, AclId.World()) }, NodeType.Ephemeral);
+ await _zk.CreateNodeAsync(parentNode + "/" + runningInfo.Address, null,
+ new List() { new Acl(AclPerm.All, AclScheme.World, AclId.World()) }, NodeType.Ephemeral);
+ await _zk.NodeExistsAsync(ZK_CLIENT_RUNNING_NODE);
+ _completionSource.SetResult(0);
+ _logger.LogInformation("Get zookeeper lock success.");
+ }
+ catch (KeeperException.NodeExistsException e)
+ {
+ if (e.getPath() != ZK_CLIENT_RUNNING_NODE)
+ {
+ _logger.LogError(e, "Error during get lock.");
+ Environment.Exit(-1);
+ }
+
+ _logger.LogInformation(
+ $"Node {ZK_CLIENT_RUNNING_NODE} exits, get Zookeeper lock failed. Other instances are running.");
+ _logger.LogWarning("Waiting...");
+ }
+ catch (Exception e)
+ {
+ _logger.LogError(e, "Exception in lock");
+ }
+ }
+
}
private async Task ConnectToZkAsync()
@@ -105,7 +175,7 @@ namespace CanalSharp.Connections
var times = 0;
while (times < 60)
{
- if (await _zk.NodeExistsAsync(ZK_RUNNING_NODE))
+ if (await _zk.NodeExistsAsync(ZK_SERVER_RUNNING_NODE))
{
return;
}
@@ -113,25 +183,36 @@ namespace CanalSharp.Connections
await Task.Delay(1000);
times++;
- _logger.LogWarning($"Can not find node {ZK_RUNNING_NODE} on Zookeeper. Retrying... {times}");
+ _logger.LogWarning($"Can not find node {ZK_SERVER_RUNNING_NODE} on Zookeeper. Retrying... {times}");
}
- throw new CanalConnectionException($"Can not find node {ZK_RUNNING_NODE} on Zookeeper.");
+ throw new CanalConnectionException($"Can not find node {ZK_SERVER_RUNNING_NODE} on Zookeeper.");
}
private async Task Zk_OnWatch(ZkWatchEventArgs args)
{
- if (args.Path == ZK_RUNNING_NODE && args.EventType == WatchEventType.NodeCreated)
+ if (args.Path == ZK_SERVER_RUNNING_NODE && args.EventType == WatchEventType.NodeCreated)
{
_serverRunningNodeReCreated = true;
- _logger.LogInformation($"Zookeeper node {ZK_RUNNING_NODE} Created");
- }
- else if (args.Path == ZK_RUNNING_NODE && args.EventType == WatchEventType.NodeDeleted)
- {
- _logger.LogInformation($"Zookeeper node {ZK_RUNNING_NODE} Deleted");
+ _logger.LogInformation($"Server node {ZK_SERVER_RUNNING_NODE} Created");
}
- await _zk.NodeExistsAsync(ZK_RUNNING_NODE);
+ if (args.Path == ZK_SERVER_RUNNING_NODE && args.EventType == WatchEventType.NodeDeleted)
+ {
+ _logger.LogInformation($"Server node {ZK_SERVER_RUNNING_NODE} Deleted");
+ }
+
+ if (args.Path == ZK_CLIENT_RUNNING_NODE && args.EventType == WatchEventType.NodeDeleted)
+ {
+ _logger.LogInformation($"Client node {ZK_SERVER_RUNNING_NODE} Deleted");
+ _ = GetZkLockAsync(_clientRunningInfo, true);
+ }
+
+ if (_zk.Connected)
+ {
+ await _zk.NodeExistsAsync(ZK_SERVER_RUNNING_NODE);
+ await _zk.NodeExistsAsync(ZK_CLIENT_RUNNING_NODE);
+ }
}
public Task SubscribeAsync(string filter = ".*\\..*")
@@ -202,7 +283,7 @@ namespace CanalSharp.Connections
return _currentConn.GetWithoutAckAsync(fetchSize, timeout);
}
- private SimpleCanalOptions CopyOptions(CanalRunningInfo runningInfo)
+ private SimpleCanalOptions CopyOptions(CanalServerRunningInfo runningInfo)
{
var tmpArr = runningInfo.Address.Split(":");
var op = new SimpleCanalOptions(tmpArr[0], int.Parse(tmpArr[1]), _options.ClientId)
diff --git a/src/CanalSharp/Connections/Simple/SimpleCanalConnection.cs b/src/CanalSharp/Connections/Simple/SimpleCanalConnection.cs
index 0a3b6c6..7d9d185 100644
--- a/src/CanalSharp/Connections/Simple/SimpleCanalConnection.cs
+++ b/src/CanalSharp/Connections/Simple/SimpleCanalConnection.cs
@@ -1,5 +1,6 @@
using System.Diagnostics.CodeAnalysis;
using System.Linq;
+using System.Net;
using System.Text;
using System.Threading.Tasks;
using BeetleX;
@@ -15,7 +16,7 @@ namespace CanalSharp.Connections
///
/// Simple connection, suitable for business scenarios with small data traffic. High Availability is not supported.
///
- public class SimpleCanalConnection : ICanalConnection
+ public class SimpleCanalConnection
{
private readonly SimpleCanalOptions _options;
private readonly ILogger _logger;
@@ -45,7 +46,6 @@ namespace CanalSharp.Connections
// _client.Socket.ReceiveTimeout = _options.SoTimeout;
//Handshake
ValidatePackageType(p, PacketType.Handshake, 1);
-
var handshake = Handshake.Parser.ParseFrom(p.Body);
var seed = handshake.Seeds;
@@ -336,6 +336,11 @@ namespace CanalSharp.Connections
return result;
}
+ internal EndPoint GetLocalEndPoint()
+ {
+ return _client.Socket.LocalEndPoint;
+ }
+
private void ValidatePackageType(Packet p, PacketType expectPacketType, int expectPacketVersion)
{
if (p.Type != expectPacketType)