Skip to content

Commit 91ef40d

Browse files
author
Elad Zelingher
committed
Initial Event-History work
1 parent 5e40fe7 commit 91ef40d

File tree

5 files changed

+185
-2
lines changed

5 files changed

+185
-2
lines changed

‎src/net45/WampSharp/WAMP2/V2/Authentication/WampCra/WampCraPendingClientDetails.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ public string AuthenticationMethod
2121

2222
public long SessionId { get; set; }
2323

24-
public string TimeStamp
24+
public string Timestamp
2525
{
2626
get
2727
{
@@ -47,7 +47,7 @@ public override string ToString()
4747
Nonce,
4848
AuthenticationProvider,
4949
AuthenticationId,
50-
TimeStamp,
50+
Timestamp,
5151
AuthenticationRole,
5252
AuthenticationMethod,
5353
SessionId);

‎src/net45/WampSharp/WAMP2/V2/MetaApi/EventHistory/IWampEventHistoryProvider.cs‎

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,149 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Reactive.Disposables;
4+
using WampSharp.Core.Serialization;
5+
using WampSharp.V2.Core;
6+
using WampSharp.V2.Core.Contracts;
7+
using WampSharp.V2.PubSub;
8+
using WampSharp.V2.Rpc;
9+
10+
namespace WampSharp.V2.MetaApi
11+
{
12+
public interface IWampEventHistoryProvider
13+
{
14+
[WampProcedure("wamp.subscription.get_events")]
15+
WampHistoricalEvent[] GetEvents(long subscriptionId, long limit = 10);
16+
}
17+
18+
public class MonitoredTopics
19+
{
20+
public string TopicUri { get; set; }
21+
22+
public SubscribeOptions SubscribeOptions { get; set; }
23+
}
24+
25+
public interface IWampEventStore
26+
{
27+
void StoreEvent<TMessage>(IWampFormatter<TMessage> formatter, long subscriptionId, WampHistoricalEvent<TMessage> historicalEvent);
28+
29+
WampHistoricalEvent<object>[] GetEvents(long subscriptionId, long limit);
30+
}
31+
32+
class WampInMemoryEventStore : IWampEventStore
33+
{
34+
public void StoreEvent<TMessage>
35+
(IWampFormatter<TMessage> formatter,
36+
long subscriptionId,
37+
WampHistoricalEvent<TMessage> historicalEvent)
38+
{
39+
}
40+
41+
public WampHistoricalEvent<object>[] GetEvents(long subscriptionId, long limit)
42+
{
43+
return new WampHistoricalEvent<object>[] {};
44+
}
45+
}
46+
47+
public class WampEventHistoryProvider : IDisposable
48+
{
49+
private readonly IWampEventStore mStore;
50+
private readonly IDisposable mDisposable;
51+
52+
public WampEventHistoryProvider(IWampTopicContainer container,
53+
IEnumerable<MonitoredTopics> topicsToRecord, IWampEventStore store)
54+
{
55+
mStore = store;
56+
57+
List<IDisposable> disposables = new List<IDisposable>();
58+
59+
foreach (MonitoredTopics topicToRecord in topicsToRecord)
60+
{
61+
EventHistorySubscriber subscriber = new EventHistorySubscriber(this);
62+
63+
IWampRegistrationSubscriptionToken disposable =
64+
container.Subscribe(subscriber,
65+
topicToRecord.TopicUri,
66+
topicToRecord.SubscribeOptions);
67+
68+
subscriber.SubscriptionId = disposable.TokenId;
69+
70+
disposables.Add(disposable);
71+
}
72+
73+
mDisposable = new CompositeDisposable(disposables);
74+
}
75+
76+
private void OnEvent<TMessage>(IWampFormatter<TMessage> formatter, long subscriptionId, WampHistoricalEvent<TMessage> wampHistoricalEvent)
77+
{
78+
mStore.StoreEvent(formatter, subscriptionId, wampHistoricalEvent);
79+
}
80+
81+
private class EventHistorySubscriber : IWampRawTopicRouterSubscriber
82+
{
83+
private readonly WampEventHistoryProvider mParent;
84+
85+
public EventHistorySubscriber(WampEventHistoryProvider parent)
86+
{
87+
mParent = parent;
88+
}
89+
90+
public long SubscriptionId
91+
{
92+
get;
93+
set;
94+
}
95+
96+
public void Event<TMessage>(IWampFormatter<TMessage> formatter,
97+
long publicationId,
98+
PublishOptions options)
99+
{
100+
InnerEvent(formatter, publicationId, options);
101+
}
102+
103+
public void Event<TMessage>(IWampFormatter<TMessage> formatter,
104+
long publicationId,
105+
PublishOptions options,
106+
TMessage[] arguments)
107+
{
108+
InnerEvent(formatter, publicationId, options, arguments);
109+
}
110+
111+
public void Event<TMessage>(IWampFormatter<TMessage> formatter,
112+
long publicationId,
113+
PublishOptions options,
114+
TMessage[] arguments,
115+
IDictionary<string, TMessage> argumentsKeywords)
116+
{
117+
InnerEvent(formatter, publicationId, options, arguments, argumentsKeywords);
118+
}
119+
120+
private void InnerEvent<TMessage>(IWampFormatter<TMessage> formatter, long publicationId, PublishOptions options, TMessage[] arguments = null, IDictionary<string, TMessage> argumentsKeywords = null)
121+
{
122+
PublishOptionsExtended optionsExtended = options as PublishOptionsExtended;
123+
124+
mParent.OnEvent(formatter,
125+
SubscriptionId,
126+
new WampHistoricalEvent<TMessage>()
127+
{
128+
Timestamp = DateTime.Now,
129+
Topic = optionsExtended.TopicUri,
130+
Arguments = arguments,
131+
ArgumentsKeywords = argumentsKeywords,
132+
PublicationId = publicationId,
133+
PublisherId = optionsExtended.PublisherId
134+
});
135+
}
136+
}
137+
138+
[WampProcedure("wamp.subscription.get_events")]
139+
public WampHistoricalEvent<object>[] GetEvents(long subscriptionId, long limit = 10)
140+
{
141+
return mStore.GetEvents(subscriptionId, limit);
142+
}
143+
144+
public void Dispose()
145+
{
146+
mDisposable.Dispose();
147+
}
148+
}
149+
}

‎src/net45/WampSharp/WAMP2/V2/MetaApi/EventHistory/WampHistoricalEvent.cs‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Runtime.Serialization;
4+
5+
namespace WampSharp.V2.MetaApi
6+
{
7+
public class WampHistoricalEvent : WampHistoricalEvent<ISerializedValue>
8+
{
9+
}
10+
11+
public class WampHistoricalEvent<TMessage>
12+
{
13+
[DataMember(Name = "timestamp")]
14+
public DateTime Timestamp { get; internal set; }
15+
16+
[DataMember(Name = "publisher")]
17+
public long PublisherId { get; internal set; }
18+
19+
[DataMember(Name = "publication")]
20+
public long PublicationId { get; internal set; }
21+
22+
[DataMember(Name = "topic")]
23+
public string Topic { get; internal set; }
24+
25+
[DataMember(Name = "args")]
26+
public TMessage[] Arguments { get; internal set; }
27+
28+
[DataMember(Name = "kwargs")]
29+
public IDictionary<string, TMessage> ArgumentsKeywords { get; internal set; }
30+
}
31+
}

‎src/net45/WampSharp/WampSharp.csproj‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -248,6 +248,8 @@
248248
<Compile Include="WAMP2\V2\MetaApi\Client\SubscriptionEvents.cs" />
249249
<Compile Include="WAMP2\V2\MetaApi\Client\WampMetaApiServiceProxy.cs" />
250250
<Compile Include="WAMP2\V2\MetaApi\Client\WampRealmProxyExtensions.cs" />
251+
<Compile Include="WAMP2\V2\MetaApi\EventHistory\IWampEventHistoryProvider.cs" />
252+
<Compile Include="WAMP2\V2\MetaApi\EventHistory\WampHistoricalEvent.cs" />
251253
<Compile Include="WAMP2\V2\MetaApi\WampHostedRealmExtensions.cs" />
252254
<Compile Include="WAMP2\V2\MetaApi\WampRealmDescriptorService.cs" />
253255
<Compile Include="WAMP2\V2\MetaApi\DescriptorServiceBase.cs" />

‎src/net45/WampSharp/WampSharp.csproj.DotSettings‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@
6868
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=wamp2_005Cv2_005Ccore_005Curi/@EntryIndexedValue">True</s:Boolean>
6969
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=wamp2_005Cv2_005Ccore_005Cv2_005Ccontracts_005Crpc/@EntryIndexedValue">True</s:Boolean>
7070
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=wamp2_005Cv2_005Cmetaapi_005Cclient/@EntryIndexedValue">True</s:Boolean>
71+
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=wamp2_005Cv2_005Cmetaapi_005Ceventhistory/@EntryIndexedValue">True</s:Boolean>
7172
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=wamp2_005Cv2_005Cmetaapi_005Cpubsub/@EntryIndexedValue">True</s:Boolean>
7273
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=wamp2_005Cv2_005Cmetaapi_005Crpc/@EntryIndexedValue">True</s:Boolean>
7374
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=wamp2_005Cv2_005Cmetaapi_005Csession/@EntryIndexedValue">True</s:Boolean>

0 commit comments

Comments
 (0)