266 lines
8.3 KiB
C#
266 lines
8.3 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Text;
|
|
using System.Threading.Tasks;
|
|
using System.Reflection;
|
|
|
|
|
|
|
|
using Google.Protobuf.WellKnownTypes;
|
|
using Google.Protobuf;
|
|
using Nettention.Proud;
|
|
using RabbitMQ.Client.Events;
|
|
using RabbitMQ.Client;
|
|
|
|
|
|
|
|
using ServerCore; using ServerBase;
|
|
|
|
|
|
namespace ServerBase;
|
|
|
|
public abstract class RabbitMQClient : RabbitMQConnectorBase, IRabbitMqSession, IInitializer
|
|
{
|
|
public delegate Task FnServerMessageRecvFromConsumer<T>(T message, BasicDeliverEventArgs ea) where T : IMessage;
|
|
public FnServerMessageRecvFromConsumer<ServerMessage>? m_fn_server_message_recv_from_consumer;
|
|
|
|
|
|
private readonly PacketReceiver m_packet_receiver;
|
|
private readonly PacketSender m_packet_sender;
|
|
|
|
|
|
public RabbitMQClient(string serviceName, string address, int port, string userName, string password, bool useSSL)
|
|
: base(serviceName, address, port, userName, password, useSSL)
|
|
{
|
|
m_packet_receiver = new PacketReceiver(this);
|
|
m_packet_sender = new PacketSender(this);
|
|
}
|
|
|
|
|
|
public virtual async Task<Result> onInit()
|
|
{
|
|
var result = new Result();
|
|
string err_msg = string.Empty;
|
|
|
|
if (false == await m_packet_receiver.registerRecvHandlerAll())
|
|
{
|
|
err_msg = $"Failed to register RecvHandler All !!! - {toBasicString()}";
|
|
result.setFail(ServerErrorCode.PacketRecvHandlerRegisterFailed, err_msg);
|
|
Log.getLogger().error(result.toBasicString());
|
|
|
|
return result;
|
|
}
|
|
|
|
if (false == await m_packet_sender.registerSendHandlerAll())
|
|
{
|
|
err_msg = $"Failed to register SendHandler All !!! - {toBasicString()}";
|
|
result.setFail(ServerErrorCode.PacketSendHandlerRegisterFailed, err_msg);
|
|
Log.getLogger().error(result.toBasicString());
|
|
|
|
return result;
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
public bool startConsumer4Client(RabbitMQClient.FnServerMessageRecvFromConsumer<ServerMessage> fnFromConsumer)
|
|
{
|
|
var err_msg = string.Empty;
|
|
|
|
if (null == fnFromConsumer)
|
|
{
|
|
err_msg = $"Param fnFromConsumers is null !!! - {toBasicString()}";
|
|
Log.getLogger().error(err_msg);
|
|
return false;
|
|
}
|
|
|
|
m_fn_server_message_recv_from_consumer = fnFromConsumer;
|
|
|
|
return base.startConsumer();
|
|
}
|
|
|
|
protected bool onRecvProtocol<T>(BasicDeliverEventArgs ea, T recvProtocol)
|
|
where T : Google.Protobuf.IMessage
|
|
{
|
|
Log.getLogger().info($"receive:{recvProtocol.ToString()} - {toBasicString()}");
|
|
|
|
|
|
var receiver = getPacketReceiver();
|
|
|
|
if (false == receiver.fillupRabbitMqPacketCommand(ea, recvProtocol, out var fillupPacketCommand))
|
|
{
|
|
Log.getLogger().error($"Invalid received Packet !!! : {recvProtocol.ToString()}");
|
|
return false;
|
|
}
|
|
|
|
if(fillupPacketCommand == null)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var found_handler = receiver.findRecvHandler(fillupPacketCommand);
|
|
if (null == found_handler)
|
|
{
|
|
Log.getLogger().error($"Unacceptable Packet !!! : {fillupPacketCommand.toBasicString()} - {toBasicString()}");
|
|
return false;
|
|
}
|
|
|
|
var task = Task.Factory.StartNew(async () =>
|
|
{
|
|
var result = await found_handler.onCheckValid(this, recvProtocol);
|
|
if (result.isFail())
|
|
{
|
|
Log.getLogger().error($"Failed to check Valid !!! : {fillupPacketCommand.toBasicString()}");
|
|
}
|
|
|
|
result = await found_handler.onProcessPacket(this, recvProtocol);
|
|
if (result.isFail())
|
|
{
|
|
Log.getLogger().error($"Failed to process Packet !!! : {fillupPacketCommand.toBasicString()}");
|
|
}
|
|
});
|
|
|
|
return true;
|
|
}
|
|
|
|
public Task<object>? sendKick( string destServer, string name, Int32 delayMS
|
|
, Action<Task<object>>? callback )
|
|
{
|
|
int reqId = nextReqId();
|
|
Task<object>? wait_task = null;
|
|
|
|
if (callback != null)
|
|
{
|
|
var cts = new CancellationTokenSource(delayMS);
|
|
|
|
wait_task = registerCompletionSource(reqId, cts.Token, callback);
|
|
if (wait_task == null)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
var message = new ServerMessage();
|
|
message.KickReq = new ServerMessage.Types.KickReq();
|
|
message.KickReq.ReqId = reqId;
|
|
message.KickReq.Name = name;
|
|
|
|
sendMessageWithJsonToChannel(destServer, message);
|
|
}
|
|
|
|
return wait_task;
|
|
}
|
|
|
|
|
|
protected override async void onRecvJsonMessageFromConsumer(object? sender, BasicDeliverEventArgs ea)
|
|
{
|
|
var body = ea.Body.ToArray();
|
|
|
|
var body_string = Encoding.UTF8.GetString(body);
|
|
var recv_message = ServerMessage.Parser.ParseJson(body_string);
|
|
|
|
if (m_fn_server_message_recv_from_consumer == null)
|
|
{
|
|
Log.getLogger().error($"m_fn_server_message_recv_from_consumer is null !!! - {toBasicString()}");
|
|
return;
|
|
}
|
|
|
|
await m_fn_server_message_recv_from_consumer.Invoke(recv_message, ea);
|
|
}
|
|
|
|
protected override void onRecvProtoMessageFromConsumer(object? sender, BasicDeliverEventArgs ea)
|
|
{
|
|
var body = ea.Body.ToArray();
|
|
|
|
var body_stream = new CodedInputStream(body);
|
|
|
|
var recv_message = new ServerMessage();
|
|
recv_message.MergeFrom(body_stream);
|
|
|
|
if (m_fn_server_message_recv_from_consumer == null)
|
|
{
|
|
Log.getLogger().error($"m_fn_server_message_recv_from_consumer is null !!! - {toBasicString()}");
|
|
return;
|
|
}
|
|
|
|
// Byte 기반 패킷 전송 테스트 후 아래 코드도 테스트 해본다.
|
|
// 이상 없다면 m_fn_server_message_recv_from_consumer 요건 제거 한다.
|
|
//onRecvProtocol(ea, recv_message);
|
|
|
|
// TODO: spooky000 - event에 async 쓸 수없어서 fire and forgot로 한다. 검토 필요.
|
|
_ = m_fn_server_message_recv_from_consumer.Invoke(recv_message, ea);
|
|
}
|
|
|
|
|
|
public void sendMessageWithByteArrayToChannel<T>(string to, T message)
|
|
where T : Google.Protobuf.IMessage
|
|
{
|
|
IConnection? con = getConnection();
|
|
if (con == null)
|
|
{
|
|
Log.getLogger().error($"Failed to sendMessageWithByteArrayToChannel() !!!, getConnetion() is null - {toBasicString()}");
|
|
return;
|
|
}
|
|
|
|
fillupSenderAndTime((message as ServerMessage));
|
|
|
|
using (var channel = con.CreateModel())
|
|
{
|
|
channel.QueueDeclare( queue: to,
|
|
durable: true,
|
|
exclusive: false,
|
|
autoDelete: false,
|
|
arguments: null);
|
|
|
|
channel.BasicPublish( exchange: "",
|
|
routingKey: to,
|
|
basicProperties: null,
|
|
body: message.ToByteArray() );
|
|
}
|
|
}
|
|
|
|
|
|
public void sendMessageWithJsonToChannel<T>(string to, T message)
|
|
where T : Google.Protobuf.IMessage
|
|
{
|
|
IConnection? con = getConnection();
|
|
if (con == null)
|
|
{
|
|
Log.getLogger().error("getConnection return null");
|
|
return;
|
|
}
|
|
|
|
fillupSenderAndTime((message as ServerMessage));
|
|
|
|
using (var channel = con.CreateModel())
|
|
{
|
|
channel.QueueDeclare( queue: to,
|
|
durable: true,
|
|
exclusive: false,
|
|
autoDelete: false,
|
|
arguments: null);
|
|
|
|
channel.BasicPublish( exchange: "",
|
|
routingKey: to,
|
|
basicProperties: null,
|
|
body: Encoding.UTF8.GetBytes(message.toJson()) );
|
|
}
|
|
}
|
|
|
|
private void fillupSenderAndTime(ServerMessage? toSendMessage)
|
|
{
|
|
if(null == toSendMessage)
|
|
{
|
|
return;
|
|
}
|
|
|
|
toSendMessage.MessageTime = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTime(DateTime.UtcNow);
|
|
toSendMessage.MessageSender = getServiceName();
|
|
}
|
|
|
|
public PacketReceiver getPacketReceiver() => m_packet_receiver;
|
|
|
|
public abstract string toBasicString();
|
|
}
|
|
|