基于surging的木舟IOT平台如何添加网络组件 - ToB企服应用市场:ToB评测及商务社交产业平台

parser.Fixed(4).Handler(
function(buffer){
var buf = BytesUtils.Slice(buffer,1,4);
parser.Fixed(buffer.ReadableBytes).Result(buf);
}).Handler(
function(buffer){parser.Fixed(8).Result(buffer);}
).Handler(function(buffer){
parser.Result('处理完成','gb2312').Complete();
}
)

复制代码

internal class TcpDeviceDataService : TcpBehavior, ITcpDeviceDataService
{
private readonly IDeviceProvider _deviceProvider;
public TcpDeviceDataService(IDeviceProvider deviceProvider)
{
_deviceProvider = deviceProvider;
}
public override void Load(string clientId, NetworkProperties tcpServerProperties)
{
var deviceStatus = _deviceProvider.IsConnected(clientId);
this.Parser.HandlePayload().Subscribe(async buffer => await ParserBuffer(buffer));
}
public override void DeviceStatusProcess(DeviceStatus status, string clientId, NetworkProperties tcpServerProperties)
{
//throw new NotImplementedException();
}
public async Task ParserBuffer(IByteBuffer buffer)
{
List<string> result = new List<string>();
while (buffer.ReadableBytes > 0)
{
result.Add(buffer.ReadString(this.Parser.GetNextFixedRecordLength(),
Encoding.GetEncoding("gb2312")));
}
// var str= buffer.ReadString(buffer.ReadableBytes, Encoding.UTF8);
var byteBuffer = Unpooled.Buffer();
byteBuffer.WriteString("\r\n", Encoding.UTF8);
byteBuffer.WriteString("处理完成", Encoding.GetEncoding("gb2312"));
await Sender.SendAndFlushAsync(byteBuffer);
// await Sender.SendAndFlushAsync("消息已接收",Encoding.GetEncoding("gb2312"));
this.Parser.Close();
}
public Task<bool> ChangeDeviceStage(string deviceId)
{
throw new NotImplementedException();
}
}

复制代码

internal class UdpDeviceDataService : UdpBehavior, IUdpDeviceDataService
{
public Task<bool> ChangeDeviceStage(string deviceId)
{
throw new NotImplementedException();
}
public override async Task Dispatch(IEnumerable<byte> bytes)
{
await Sender.SendAndFlushAsync("\r\n", Encoding.UTF8);
await Sender.SendAndFlushAsync("处理完成", Encoding.GetEncoding("gb2312"));
}
}

复制代码

internal class WSDeviceDataService : WSBehavior, IWSDeviceDataService
{
protected override void OnMessage(MessageEventArgs e)
{
this.Client.Value.SendTo($"send:{e.Data},\r\n reply:hello,welcome to you!",ID);
}
protected override void OnOpen()
{
}
}

复制代码

syntax = "proto3";
package Greet;
service Greeter {
// Sends a greeting
rpc ChangeDeviceStage (DeviceRequest) returns (DeviceReply) {}
}
message DeviceRequest {
string deviceId = 1;
}
message DeviceReply {
bool message = 1;
}

复制代码

public partial class GreeterBehavior : Greeter.GreeterBase, IServiceBehavior
{
private ServerReceivedDelegate received;
public event ServerReceivedDelegate Received
{
add
{
if (value == null)
{
received += value;
}
}
remove
{
received -= value;
}
}
public string MessageId { get; } = Guid.NewGuid().ToString("N");
public async Task Write(object result, int statusCode = 200, string exceptionMessage = "")
{
if (received == null)
return;
var message = new TransportMessage(MessageId, new ReactiveResultMessage
{
ExceptionMessage = exceptionMessage,
StatusCode = statusCode,
Result = result
});
await received(message);
}
public T CreateProxy<T>(string key) where T : class
{
return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy<T>(key);
}
public object CreateProxy(Type type)
{
return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy(type);
}
public object CreateProxy(string key, Type type)
{
return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy(key, type);
}
public T CreateProxy<T>() where T : class
{
return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy<T>();
}
public T GetService<T>(string key) where T : class
{
if (ServiceLocator.Current.IsRegisteredWithKey<T>(key))
return ServiceLocator.GetService<T>(key);
else
return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy<T>(key);
}
public T GetService<T>() where T : class
{
if (ServiceLocator.Current.IsRegistered<T>())
return ServiceLocator.GetService<T>();
else
return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy<T>();
}
public object GetService(Type type)
{
if (ServiceLocator.Current.IsRegistered(type))
return ServiceLocator.GetService(type);
else
return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy(type);
}
public object GetService(string key, Type type)
{
if (ServiceLocator.Current.IsRegisteredWithKey(key, type))
return ServiceLocator.GetService(key, type);
else
return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy(key, type);
}
public void Publish(IntegrationEvent @event)
{
GetService<IEventBus>().Publish(@event);
}
}

复制代码

public class GrpcDeviceDataService : GreeterBehavior, IGrpcDeviceDataService
{
public override Task<DeviceReply> ChangeDeviceStage(DeviceRequest request, ServerCallContext context)
{
return Task.FromResult(new DeviceReply
{
Message = true
}) ;
}
}

复制代码

public class MQTTDeviceDataService : MqttBehavior, IMQTTDeviceDataService
{
public override async Task<bool> Authorized(string username, string password)
{
bool result = false;
if (username == "admin" && password == "123456")
result = true;
return await Task.FromResult(result);
}
public async Task<bool> IsOnline(string deviceId)
{
return await base.GetDeviceIsOnine(deviceId);
}
public async Task Publish(string deviceId, WillMessage message)
{
var willMessage = new MqttWillMessage
{
WillMessage = message.Message,
Qos = message.Qos,
Topic = message.Topic,
WillRetain = message.WillRetain
};
await Publish(deviceId, willMessage);
await RemotePublish(deviceId, willMessage);
}
}

复制代码