forked from planetarium/libplanet
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: implement PubsubSwarm mode for Node
- Loading branch information
Showing
19 changed files
with
675 additions
and
13 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| using System.ComponentModel; | ||
| using Libplanet.Node.DataAnnotations; | ||
|
|
||
| namespace Libplanet.Node.Options; | ||
|
|
||
| [Options(Position)] | ||
| public class PubsubSwarmOptions : OptionsBase<PubsubSwarmOptions> | ||
| { | ||
| public const string Position = "PubsubSwarm"; | ||
|
|
||
| public bool Proposer { get; set; } | ||
|
|
||
| public long BlockInterval { get; set; } = 4000; | ||
|
|
||
| [PrivateKey] | ||
| [Description("The private key of the node.")] | ||
| public string PrivateKey { get; set; } = string.Empty; | ||
| } |
21 changes: 21 additions & 0 deletions
21
sdk/node/Libplanet.Node/Options/PubsubSwarmOptionsConfigurator.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| using Libplanet.Common; | ||
| using Libplanet.Crypto; | ||
| using Microsoft.Extensions.Logging; | ||
|
|
||
| namespace Libplanet.Node.Options; | ||
|
|
||
| internal sealed class PubsubSwarmOptionsConfigurator( | ||
| ILogger<PubsubSwarmOptionsConfigurator> logger) | ||
| : OptionsConfiguratorBase<PubsubSwarmOptions> | ||
| { | ||
| protected override void OnConfigure(PubsubSwarmOptions options) | ||
| { | ||
| if (options.PrivateKey == string.Empty) | ||
| { | ||
| options.PrivateKey = ByteUtil.Hex(new PrivateKey().ByteArray); | ||
| logger.LogWarning( | ||
| "Node's private key is not set. A new private key is generated: {PrivateKey}", | ||
| options.PrivateKey); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| using Libplanet.Net.Messages; | ||
|
|
||
| namespace Libplanet.Node.Protocols; | ||
|
|
||
| public static class MessageContentCodec | ||
| { | ||
| public static MessageContent Deserialize(byte[] bytes) | ||
| { | ||
| var rawFrames = bytes; | ||
| var type = rawFrames[0]; | ||
| rawFrames = rawFrames.Skip(1).ToArray(); | ||
| var dataFrames = new List<byte[]>(); | ||
| var frameCount = BitConverter.ToInt32(rawFrames[..4]); | ||
| rawFrames = rawFrames.Skip(4).ToArray(); | ||
| for (int i = 0; i < frameCount; i++) | ||
| { | ||
| var frameSize = BitConverter.ToInt32(rawFrames[..4]); | ||
| dataFrames.Add(rawFrames.Skip(4).Take(frameSize).ToArray()); | ||
| rawFrames = rawFrames.Skip(4 + frameSize).ToArray(); | ||
| } | ||
|
|
||
| return NetMQMessageCodec.CreateMessage( | ||
| (MessageContent.MessageType)type, | ||
| dataFrames.ToArray()); | ||
| } | ||
|
|
||
| public static byte[] Serialize(MessageContent messageContent) | ||
| { | ||
| IEnumerable<byte> ba = [(byte)messageContent.Type]; | ||
| ba = ba.Concat(BitConverter.GetBytes(messageContent.DataFrames.Count())); | ||
| foreach (var bytes in messageContent.DataFrames) | ||
| { | ||
| ba = ba.Concat(BitConverter.GetBytes(bytes.Length)); | ||
| ba = ba.Concat(bytes); | ||
| } | ||
|
|
||
| return ba.ToArray(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| using System.Buffers; | ||
| using Libplanet.Node.Services; | ||
| using Nethermind.Libp2p.Core; | ||
|
|
||
| namespace Libplanet.Node.Protocols | ||
| { | ||
| internal class PingPongProtocol(IMessageChannel messageChannel) : IProtocol | ||
| { | ||
| public string Id => "/blockchain/ping-pong/1.0.0"; | ||
|
|
||
| public async Task DialAsync( | ||
| IChannel downChannel, | ||
| IChannelFactory? upChannelFactory, | ||
| IPeerContext context) | ||
| { | ||
| TimeSpan elapsed = TimeSpan.Zero; | ||
|
|
||
| _ = Task.Run( | ||
| async () => | ||
| { | ||
| while (true) | ||
| { | ||
| ReadOnlySequence<byte> read = | ||
| await downChannel.ReadAsync(0, ReadBlockingMode.WaitAny).OrThrow(); | ||
| messageChannel.ReceiveMessage( | ||
| context.RemotePeer.Address, | ||
| read.ToArray(), | ||
| reply => _ = | ||
| downChannel.WriteAsync( | ||
| new ReadOnlySequence<byte>(reply))); | ||
| } | ||
| }); | ||
|
|
||
| // Keep connection for 5 seconds | ||
| while (true) | ||
| { | ||
| if (messageChannel.TryGetMessageToSend( | ||
| context.RemotePeer.Address, | ||
| out byte[]? msg)) | ||
| { | ||
| if (msg is not null) | ||
| { | ||
| await downChannel.WriteAsync( | ||
| new ReadOnlySequence<byte>(msg)); | ||
| } | ||
|
|
||
| elapsed = TimeSpan.Zero; | ||
| } | ||
| else | ||
| { | ||
| await Task.Delay(50); | ||
| elapsed += TimeSpan.FromMilliseconds(50); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public async Task ListenAsync( | ||
| IChannel downChannel, | ||
| IChannelFactory? upChannelFactory, | ||
| IPeerContext context) | ||
| { | ||
| while (true) | ||
| { | ||
| ReadOnlySequence<byte> read = | ||
| await downChannel.ReadAsync(0, ReadBlockingMode.WaitAny).OrThrow(); | ||
| messageChannel.ReceiveMessage( | ||
| context.RemotePeer.Address, | ||
| read.ToArray(), | ||
| msg => _ = | ||
| downChannel.WriteAsync( | ||
| new ReadOnlySequence<byte>(msg))); | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| using Libplanet.Net.Messages; | ||
| using Libplanet.Node.Protocols; | ||
| using Multiformats.Address; | ||
|
|
||
| namespace Libplanet.Node.Services; | ||
|
|
||
| public interface IMessageChannel | ||
| { | ||
| EventHandler<(Multiaddress, byte[], Action<byte[]>)>? OnMessageReceived | ||
| { | ||
| get; | ||
| set; | ||
| } | ||
|
|
||
| void SendMessage(Multiaddress address, byte[] message); | ||
|
|
||
| void ReceiveMessage( | ||
| Multiaddress sender, | ||
| byte[] message, | ||
| Action<byte[]> callback); | ||
|
|
||
| bool TryGetMessageToSend(Multiaddress address, out byte[]? message); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| // SPDX-FileCopyrightText: 2024 Demerzel Solutions Limited | ||
| // SPDX-License-Identifier: MIT | ||
|
|
||
| using System.Collections.Concurrent; | ||
| using Multiformats.Address; | ||
|
|
||
| namespace Libplanet.Node.Services; | ||
|
|
||
| public class MessageChannel : IMessageChannel | ||
| { | ||
| private readonly ConcurrentDictionary<Multiaddress, ConcurrentBag<byte[]>> | ||
| _messagesToSend = new(); | ||
|
|
||
| public EventHandler<(Multiaddress, byte[], Action<byte[]>)>? OnMessageReceived | ||
| { | ||
| get; | ||
| set; | ||
| } | ||
|
|
||
| public void SendMessage(Multiaddress address, byte[] message) | ||
| { | ||
| if (!_messagesToSend.TryGetValue(address, out ConcurrentBag<byte[]>? bag)) | ||
| { | ||
| bag = []; | ||
| _messagesToSend.TryAdd(address, bag); | ||
| } | ||
|
|
||
| bag.Add(message); | ||
| } | ||
|
|
||
| public void ReceiveMessage( | ||
| Multiaddress sender, | ||
| byte[] message, | ||
| Action<byte[]> callback) | ||
| { | ||
| OnMessageReceived?.Invoke(this, (sender, message, callback)); | ||
| } | ||
|
|
||
| public bool TryGetMessageToSend(Multiaddress address, out byte[]? message) | ||
| { | ||
| if (_messagesToSend.TryGetValue(address, out ConcurrentBag<byte[]>? bag)) | ||
| { | ||
| if (bag.IsEmpty) | ||
| { | ||
| message = null; | ||
| return false; | ||
| } | ||
|
|
||
| return bag.TryTake(out message); | ||
| } | ||
|
|
||
| message = null; | ||
| return false; | ||
| } | ||
| } |
Oops, something went wrong.