Skip to content

Commit a6df889

Browse files
fanliang11fanliang11
committed
support reactive programming
1 parent 3037b35 commit a6df889

17 files changed

Lines changed: 382 additions & 11 deletions

File tree

‎src/Surging.Core/Surging.Core.CPlatform/Ioc/IServiceBehavior.cs‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,15 @@
1-
using System;
1+
using Surging.Core.CPlatform.Messages;
2+
using System;
23
using System.Collections.Generic;
34
using System.Text;
5+
using System.Threading.Tasks;
46

57
namespace Surging.Core.CPlatform.Ioc
68
{
7-
public interface IServiceBehavior
9+
public delegate Task ServerReceivedDelegate(TransportMessage message);
10+
public interface IServiceBehavior
811
{
12+
public string MessageId { get; }
13+
event ServerReceivedDelegate Received;
914
}
1015
}

‎src/Surging.Core/Surging.Core.CPlatform/Ioc/ServiceBase.cs‎

Lines changed: 33 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,44 @@
1-
using Surging.Core.CPlatform.Utilities;
1+
using Surging.Core.CPlatform.Messages;
2+
using Surging.Core.CPlatform.Utilities;
23
using System;
34
using System.Collections.Generic;
45
using System.Text;
6+
using System.Threading.Tasks;
57

68
namespace Surging.Core.CPlatform.Ioc
79
{
810
public abstract class ServiceBase: IServiceBehavior
911
{
12+
private ServerReceivedDelegate received;
13+
public event ServerReceivedDelegate Received
14+
{
15+
add
16+
{
17+
if (received == null)
18+
{
19+
received += value;
20+
}
21+
}
22+
remove
23+
{
24+
received -= value;
25+
}
26+
}
27+
28+
public string MessageId { get; } = Guid.NewGuid().ToString("N");
29+
public async Task Write(object result, int statusCode = 200, string exceptionMessage = "")
30+
{
31+
if (received == null)
32+
return;
33+
var message = new TransportMessage(MessageId, new ReactiveResultMessage
34+
{
35+
ExceptionMessage = exceptionMessage,
36+
StatusCode = statusCode,
37+
Result = result
38+
39+
});
40+
await received(message);
41+
}
1042
public virtual T GetService<T>() where T : class
1143
{
1244
return ServiceLocator.GetService<T>();

‎src/Surging.Core/Surging.Core.CPlatform/Messages/MessagePackTransportMessageType.cs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,5 +14,7 @@ public class MessagePackTransportMessageType
1414
public static string httpMessageTypeName = typeof(HttpMessage).FullName;
1515

1616
public static string httpResultMessageTypeName = typeof(HttpResultMessage<object>).FullName;
17+
18+
public static string reactiveResultMessageTypeName = typeof(ReactiveResultMessage).FullName;
1719
}
1820
}

‎src/Surging.Core/Surging.Core.CPlatform/Messages/ReactiveResultMessage.cs‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Linq;
4+
using System.Text;
5+
using System.Threading.Tasks;
6+
7+
namespace Surging.Core.CPlatform.Messages
8+
{
9+
public class ReactiveResultMessage
10+
{
11+
/// <summary>
12+
/// 异常消息。
13+
/// </summary>
14+
public string ExceptionMessage { get; set; }
15+
16+
/// <summary>
17+
/// 状态码
18+
/// </summary>
19+
public int StatusCode { get; set; } = 200;
20+
/// <summary>
21+
/// 结果内容。
22+
/// </summary>
23+
public object Result { get; set; }
24+
}
25+
}

‎src/Surging.Core/Surging.Core.CPlatform/Messages/TransportMessage.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,12 @@ public bool IsInvokeResultMessage()
7878
return ContentType == MessagePackTransportMessageType.remoteInvokeResultMessageTypeName;
7979
}
8080

81+
[MethodImpl(MethodImplOptions.AggressiveInlining)]
82+
public bool IsReactiveMessage()
83+
{
84+
return ContentType == MessagePackTransportMessageType.reactiveResultMessageTypeName;
85+
}
86+
8187
[MethodImpl(MethodImplOptions.AggressiveInlining)]
8288
public bool IsHttpMessage()
8389
{

‎src/Surging.Core/Surging.Core.CPlatform/Runtime/Server/Implementation/ServiceDiscovery/Attributes/ReactiveAttribute.cs‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Linq;
4+
using System.Text;
5+
using System.Threading.Tasks;
6+
7+
namespace Surging.Core.CPlatform.Runtime.Server.Implementation.ServiceDiscovery.Attributes
8+
{
9+
[AttributeUsage(AttributeTargets.Method, AllowMultiple = true, Inherited = true)]
10+
public class ReactiveAttribute : Attribute
11+
{
12+
}
13+
}

‎src/Surging.Core/Surging.Core.CPlatform/Runtime/Server/Implementation/ServiceDiscovery/Implementation/ClrServiceEntryFactory.cs‎

Lines changed: 65 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,11 @@
1414
using Surging.Core.CPlatform.Validation;
1515
using static Surging.Core.CPlatform.Utilities.FastInvoke;
1616
using System.Threading;
17+
using System.Runtime.CompilerServices;
18+
using Surging.Core.CPlatform.Messages;
19+
using Surging.Core.CPlatform.Exceptions;
20+
using System.Collections.Concurrent;
21+
using Surging.Core.CPlatform.Ioc;
1722

1823
namespace Surging.Core.CPlatform.Runtime.Server.Implementation.ServiceDiscovery.Implementation
1924
{
@@ -27,7 +32,8 @@ public class ClrServiceEntryFactory : IClrServiceEntryFactory
2732
private readonly IServiceIdGenerator _serviceIdGenerator;
2833
private readonly ITypeConvertibleService _typeConvertibleService;
2934
private readonly IValidationProcessor _validationProcessor;
30-
35+
private readonly ConcurrentDictionary<string, ManualResetValueTaskSource<TransportMessage>> _resultDictionary =
36+
new ConcurrentDictionary<string, ManualResetValueTaskSource<TransportMessage>>();
3137
#endregion Field
3238

3339
#region Constructor
@@ -106,8 +112,8 @@ private ServiceEntry Create(MethodInfo method, string serviceName, string routeT
106112
var fastInvoker = GetHandler(serviceId, method);
107113

108114
var methodValidateAttribute = attributes.Where(p => p is ValidateAttribute)
109-
.Cast<ValidateAttribute>().FirstOrDefault();
110-
115+
.Cast<ValidateAttribute>().FirstOrDefault();
116+
var isReactive = attributes.Any(p => p is ReactiveAttribute);
111117
return new ServiceEntry
112118
{
113119
Descriptor = serviceDescriptor,
@@ -117,7 +123,7 @@ private ServiceEntry Create(MethodInfo method, string serviceName, string routeT
117123
Parameters = method.GetParameters(),
118124
Type = method.DeclaringType,
119125
Attributes = attributes,
120-
Func = (key, parameters) =>
126+
Func = async (key, parameters) =>
121127
{
122128
object instance = null;
123129
if (AppConfig.ServerOptions.IsModulePerLifetimeScope)
@@ -148,12 +154,64 @@ private ServiceEntry Create(MethodInfo method, string serviceName, string routeT
148154
var parameter = _typeConvertibleService.Convert(value, parameterType);
149155
list.Add(parameter);
150156
}
151-
var result = fastInvoker(instance, list.ToArray());
152-
return Task.FromResult(result);
157+
if (!isReactive)
158+
{
159+
var result = fastInvoker(instance, list.ToArray());
160+
return await Task.FromResult(result);
161+
}
162+
else
163+
{
164+
var serviceBehavior = instance as IServiceBehavior;
165+
var callbackTask = RegisterResultCallbackAsync(serviceBehavior.MessageId, Task.Factory.CancellationToken);
166+
serviceBehavior.Received += MessageListener_Received;
167+
var result = fastInvoker(instance, list.ToArray());
168+
return await callbackTask;
169+
}
153170
}
154171
};
155172
}
156-
173+
174+
[MethodImpl(MethodImplOptions.AggressiveInlining)]
175+
private async Task<object> RegisterResultCallbackAsync(string id, CancellationToken cancellationToken)
176+
{
177+
178+
var task = new ManualResetValueTaskSource<TransportMessage>();
179+
_resultDictionary.TryAdd(id, task);
180+
try
181+
{
182+
var result = await task.AwaitValue(cancellationToken);
183+
return result.GetContent<ReactiveResultMessage>()?.Result;
184+
}
185+
finally
186+
{
187+
//删除回调任务
188+
ManualResetValueTaskSource<TransportMessage> value;
189+
_resultDictionary.TryRemove(id, out value);
190+
value.SetCanceled();
191+
}
192+
}
193+
194+
[MethodImpl(MethodImplOptions.AggressiveInlining)]
195+
private async Task MessageListener_Received(TransportMessage message)
196+
{
197+
ManualResetValueTaskSource<TransportMessage> task;
198+
if (!_resultDictionary.TryGetValue(message.Id, out task))
199+
return;
200+
201+
if (message.IsReactiveMessage())
202+
{
203+
var content = message.GetContent<ReactiveResultMessage>();
204+
if (!string.IsNullOrEmpty(content.ExceptionMessage))
205+
{
206+
task.SetException(new CPlatformCommunicationException(content.ExceptionMessage, content.StatusCode));
207+
}
208+
else
209+
{
210+
task.SetResult(message);
211+
}
212+
}
213+
}
214+
157215
private FastInvokeHandler GetHandler(string key, MethodInfo method)
158216
{
159217
var objInstance = ServiceResolver.Current.GetService(null, key);

‎src/Surging.Core/Surging.Core.DNS/Runtime/DnsBehavior.cs‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,11 +11,42 @@
1111
using Surging.Core.CPlatform.EventBus.Events;
1212
using Surging.Core.CPlatform.Module;
1313
using System.Threading.Tasks;
14+
using Surging.Core.CPlatform.Messages;
1415

1516
namespace Surging.Core.DNS.Runtime
1617
{
1718
public abstract class DnsBehavior : IServiceBehavior
1819
{
20+
private ServerReceivedDelegate received;
21+
public event ServerReceivedDelegate Received
22+
{
23+
add
24+
{
25+
if (received == null)
26+
{
27+
received += value;
28+
}
29+
}
30+
remove
31+
{
32+
received -= value;
33+
}
34+
}
35+
36+
public string MessageId { get; } = Guid.NewGuid().ToString("N");
37+
public async Task Write(object result, int statusCode = 200, string exceptionMessage = "")
38+
{
39+
if (received == null)
40+
return;
41+
var message = new TransportMessage(MessageId, new ReactiveResultMessage
42+
{
43+
ExceptionMessage = exceptionMessage,
44+
StatusCode = statusCode,
45+
Result = result
46+
47+
});
48+
await received(message);
49+
}
1950
public T CreateProxy<T>(string key) where T : class
2051
{
2152
return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy<T>(key);

‎src/Surging.Core/Surging.Core.DotNettyWSServer/Runtime/WSBehavior.cs‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
using Surging.Core.CPlatform.EventBus.Events;
44
using Surging.Core.CPlatform.EventBus.Implementation;
55
using Surging.Core.CPlatform.Ioc;
6+
using Surging.Core.CPlatform.Messages;
67
using Surging.Core.CPlatform.Utilities;
78
using Surging.Core.ProxyGenerator;
89
using System;
@@ -14,6 +15,36 @@ namespace Surging.Core.DotNettyWSServer.Runtime
1415
{
1516
public class WSBehavior : IServiceBehavior
1617
{
18+
private ServerReceivedDelegate received;
19+
public event ServerReceivedDelegate Received
20+
{
21+
add
22+
{
23+
if (received == null)
24+
{
25+
received += value;
26+
}
27+
}
28+
remove
29+
{
30+
received -= value;
31+
}
32+
}
33+
34+
public string MessageId { get; } = Guid.NewGuid().ToString("N");
35+
public async Task Write(object result, int statusCode = 200, string exceptionMessage = "")
36+
{
37+
if (received == null)
38+
return;
39+
var message = new TransportMessage(MessageId, new ReactiveResultMessage
40+
{
41+
ExceptionMessage = exceptionMessage,
42+
StatusCode = statusCode,
43+
Result = result
44+
45+
});
46+
await received(message);
47+
}
1748
public T CreateProxy<T>(string key) where T : class
1849
{
1950
return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy<T>(key);

‎src/Surging.Core/Surging.Core.Protocol.Udp/Runtime/UdpBehavior.cs‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
using Surging.Core.CPlatform.EventBus.Events;
33
using Surging.Core.CPlatform.EventBus.Implementation;
44
using Surging.Core.CPlatform.Ioc;
5+
using Surging.Core.CPlatform.Messages;
56
using Surging.Core.CPlatform.Utilities;
67
using Surging.Core.ProxyGenerator;
78
using System;
@@ -13,6 +14,36 @@ namespace Surging.Core.Protocol.Udp.Runtime
1314
{
1415
public abstract class UdpBehavior : IServiceBehavior
1516
{
17+
private ServerReceivedDelegate received;
18+
public event ServerReceivedDelegate Received
19+
{
20+
add
21+
{
22+
if (received == null)
23+
{
24+
received += value;
25+
}
26+
}
27+
remove
28+
{
29+
received -= value;
30+
}
31+
}
32+
33+
public string MessageId { get; } = Guid.NewGuid().ToString("N");
34+
public async Task Write(object result, int statusCode = 200, string exceptionMessage = "")
35+
{
36+
if (received == null)
37+
return;
38+
var message = new TransportMessage(MessageId, new ReactiveResultMessage
39+
{
40+
ExceptionMessage = exceptionMessage,
41+
StatusCode = statusCode,
42+
Result = result
43+
44+
});
45+
await received(message);
46+
}
1647
public T CreateProxy<T>(string key) where T : class
1748
{
1849
return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy<T>(key);

0 commit comments

Comments
 (0)