1+ //-----------------------------------------------------------------------
2+ // <copyright file="DurableShardingSpec.cs" company="Akka.NET Project">
3+ // Copyright (C) 2009-2023 Lightbend Inc. <http://www.lightbend.com>
4+ // Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
5+ // </copyright>
6+ //-----------------------------------------------------------------------
7+
8+ using System ;
9+ using System . Threading . Tasks ;
10+ using Akka . Actor ;
11+ using Akka . Actor . Dsl ;
12+ using Akka . Cluster . Sharding . Delivery ;
13+ using Akka . Configuration ;
14+ using Akka . Delivery ;
15+ using Akka . Event ;
16+ using Akka . Persistence . Delivery ;
17+ using Akka . TestKit ;
18+ using Xunit ;
19+ using Xunit . Abstractions ;
20+ using FluentAssertions ;
21+ using static Akka . Tests . Delivery . TestConsumer ;
22+
23+ namespace Akka . Cluster . Sharding . Tests . Delivery ;
24+
25+ public class DurableShardingSpec : AkkaSpec
26+ {
27+ public static readonly Config Config = @"
28+ akka.actor.provider = cluster
29+ akka.remote.dot-netty.tcp.port = 0
30+ akka.reliable-delivery.consumer-controller.flow-control-window = 20
31+ akka.persistence.journal.plugin = ""akka.persistence.journal.inmem""
32+ akka.persistence.snapshot-store.plugin = ""akka.persistence.snapshot-store.inmem""
33+ " ;
34+
35+ public DurableShardingSpec ( ITestOutputHelper output ) : base ( Config , output )
36+ {
37+ // TODO: add journal operations subscriptions, once that's properly supported in Akka.Persistence
38+ }
39+
40+ private int _idCount ;
41+
42+ private string ProducerId => $ "p-{ _idCount } ";
43+
44+ private int NextId ( )
45+ {
46+ return _idCount ++ ;
47+ }
48+
49+ private async Task JoinCluster ( )
50+ {
51+ var cluster = Cluster . Get ( Sys ) ;
52+ await cluster . JoinAsync ( cluster . SelfAddress ) ;
53+ await AwaitAssertAsync ( ( ) => Assert . True ( cluster . IsUp ) ) ;
54+ }
55+
56+ [ Fact ]
57+ public async Task ReliableDelivery_with_sharding_and_durable_queue_must_load_initial_state_and_resend_unconfirmed ( )
58+ {
59+ await JoinCluster ( ) ;
60+ NextId ( ) ;
61+
62+ var consumerProbe = CreateTestProbe ( ) ;
63+ var sharding = await ClusterSharding . Get ( Sys ) . StartAsync ( $ "TestConsumer-{ _idCount } ", s =>
64+ ShardingConsumerController . Create < Job > ( c =>
65+ Props . Create ( ( ) => new Consumer ( c , consumerProbe ) ) ,
66+ ShardingConsumerController . Settings . Create ( Sys ) ) , ClusterShardingSettings . Create ( Sys ) ,
67+ HashCodeMessageExtractor . Create ( 10 ,
68+ o =>
69+ {
70+ if ( o is ShardingEnvelope se )
71+ return se . EntityId ;
72+ return string . Empty ;
73+ } , o =>
74+ {
75+ if ( o is ShardingEnvelope se )
76+ return se . Message ;
77+ return o ;
78+ } ) ) ;
79+
80+ var durableQueueProps = EventSourcedProducerQueue . Create < Job > ( ProducerId , Sys ) ;
81+ var shardingProducerController =
82+ Sys . ActorOf (
83+ ShardingProducerController . Create < Job > ( ProducerId , sharding , durableQueueProps ,
84+ ShardingProducerController . Settings . Create ( Sys ) ) , $ "shardingProducerController-{ _idCount } ") ;
85+ var producerProbe = CreateTestProbe ( ) ;
86+ shardingProducerController . Tell ( new ShardingProducerController . Start < Job > ( producerProbe . Ref ) ) ;
87+
88+ for ( var i = 1 ; i <= 4 ; i ++ )
89+ {
90+ ( await producerProbe . ExpectMsgAsync < ShardingProducerController . RequestNext < Job > > ( ) ) . SendNextTo . Tell (
91+ new ShardingEnvelope ( "entity-1" , new Job ( $ "msg-{ i } ") ) ) ;
92+ // TODO: need journal operations queries here to verify that the message was persisted
93+ }
94+
95+ var delivery1 = await consumerProbe . ExpectMsgAsync < JobDelivery > ( ) ;
96+ delivery1 . ConfirmTo . Tell ( ConsumerController . Confirmed . Instance ) ;
97+ // TODO: need journal operations queries here to verify that the Confirmed was persisted
98+
99+ var delivery2 = await consumerProbe . ExpectMsgAsync < JobDelivery > ( ) ;
100+ delivery2 . ConfirmTo . Tell ( ConsumerController . Confirmed . Instance ) ;
101+ // TODO: need journal operations queries here to verify that the Confirmed was persisted
102+
103+ await producerProbe . ExpectMsgAsync < ShardingProducerController . RequestNext < Job > > ( ) ;
104+
105+ // let the initial messages reach the ShardingConsumerController before stopping ShardingProducerController
106+ var delivery3 = await consumerProbe . ExpectMsgAsync < JobDelivery > ( ) ;
107+ delivery3 . Msg . Should ( ) . Be ( new Job ( "msg-3" ) ) ;
108+ delivery3 . SeqNr . Should ( ) . Be ( 3 ) ;
109+
110+ await Task . Delay ( 1000 ) ;
111+
112+ Sys . Log . Info ( "Stopping [{0}]" , shardingProducerController ) ;
113+ Watch ( shardingProducerController ) ;
114+ Sys . Stop ( shardingProducerController ) ;
115+ await ExpectTerminatedAsync ( shardingProducerController ) ;
116+
117+ var shardingProducerController2 =
118+ Sys . ActorOf (
119+ ShardingProducerController . Create < Job > ( ProducerId , sharding , durableQueueProps ,
120+ ShardingProducerController . Settings . Create ( Sys ) ) , $ "shardingProducerController2-{ _idCount } ") ;
121+ shardingProducerController2 . Tell ( new ShardingProducerController . Start < Job > ( producerProbe . Ref ) ) ;
122+
123+ // delivery3 and delivery4 are still from old shardingProducerController, that were queued in ConsumerController
124+ delivery3 . ConfirmTo . Tell ( ConsumerController . Confirmed . Instance ) ;
125+ // that confirmation goes to old dead shardingProducerController, and therefore not stored
126+ // TODO: need journal operations queries here to verify that the Confirmed WAS NOT persisted
127+
128+ var delivery4 = await consumerProbe . ExpectMsgAsync < JobDelivery > ( ) ;
129+ delivery4 . Msg . Should ( ) . Be ( new Job ( "msg-4" ) ) ;
130+ delivery4 . SeqNr . Should ( ) . Be ( 4 ) ;
131+ delivery4 . ConfirmTo . Tell ( ConsumerController . Confirmed . Instance ) ;
132+ // that confirmation goes to old dead shardingProducerController, and therefore not stored
133+ // TODO: need journal operations queries here to verify that the Confirmed WAS NOT persisted
134+
135+ // now the unconfirmed are re-delivered
136+ var redelivery3 = await consumerProbe . ExpectMsgAsync < JobDelivery > ( ) ;
137+ redelivery3 . Msg . Should ( ) . Be ( new Job ( "msg-3" ) ) ;
138+ redelivery3 . SeqNr . Should ( ) . Be ( 1 ) ; // new ProducerController and there starting at 1
139+ redelivery3 . ConfirmTo . Tell ( ConsumerController . Confirmed . Instance ) ;
140+ // TODO: need journal operations queries here to verify that the Confirmed was persisted
141+
142+ var redelivery4 = await consumerProbe . ExpectMsgAsync < JobDelivery > ( ) ;
143+ redelivery4 . Msg . Should ( ) . Be ( new Job ( "msg-4" ) ) ;
144+ redelivery4 . SeqNr . Should ( ) . Be ( 2 ) ;
145+ redelivery4 . ConfirmTo . Tell ( ConsumerController . Confirmed . Instance ) ;
146+ // TODO: need journal operations queries here to verify that the Confirmed was persisted
147+
148+ var next5 = await producerProbe . ExpectMsgAsync < ShardingProducerController . RequestNext < Job > > ( ) ;
149+ next5 . SendNextTo . Tell ( new ShardingEnvelope ( "entity-1" , new Job ( "msg-5" ) ) ) ;
150+ // TODO: need journal operations queries here to verify that the message was persisted
151+
152+
153+ // the consumer controller may have stopped after msg-5, so allow for resend on timeout (10-15s)
154+ var delivery5 = await consumerProbe . ExpectMsgAsync < JobDelivery > ( TimeSpan . FromSeconds ( 20 ) ) ;
155+ delivery5 . Msg . Should ( ) . Be ( new Job ( "msg-5" ) ) ;
156+ delivery5 . SeqNr . Should ( ) . Be ( 3 ) ; // 3, instead of 5, because SeqNr reset upon ProducerController restart
157+ delivery5 . ConfirmTo . Tell ( ConsumerController . Confirmed . Instance ) ;
158+ // TODO: need journal operations queries here to verify that the Confirmed was persisted
159+ }
160+
161+ [ Fact ]
162+ public async Task ReliableDelivery_with_sharding_and_durable_queue_must_reply_to_MessageWithConfirmation_after_storage ( )
163+ {
164+ await JoinCluster ( ) ;
165+ NextId ( ) ;
166+
167+ var consumerProbe = CreateTestProbe ( ) ;
168+ var sharding = await ClusterSharding . Get ( Sys ) . StartAsync ( $ "TestConsumer-{ _idCount } ", s =>
169+ ShardingConsumerController . Create < Job > ( c =>
170+ Props . Create ( ( ) => new Consumer ( c , consumerProbe ) ) ,
171+ ShardingConsumerController . Settings . Create ( Sys ) ) , ClusterShardingSettings . Create ( Sys ) ,
172+ HashCodeMessageExtractor . Create ( 10 ,
173+ o =>
174+ {
175+ if ( o is ShardingEnvelope se )
176+ return se . EntityId ;
177+ return string . Empty ;
178+ } , o =>
179+ {
180+ if ( o is ShardingEnvelope se )
181+ return se . Message ;
182+ return o ;
183+ } ) ) ;
184+
185+ var durableQueueProps = EventSourcedProducerQueue . Create < Job > ( ProducerId , Sys ) ;
186+ var shardingProducerController =
187+ Sys . ActorOf (
188+ ShardingProducerController . Create < Job > ( ProducerId , sharding , durableQueueProps ,
189+ ShardingProducerController . Settings . Create ( Sys ) ) , $ "shardingProducerController-{ _idCount } ") ;
190+ var producerProbe = CreateTestProbe ( ) ;
191+ shardingProducerController . Tell ( new ShardingProducerController . Start < Job > ( producerProbe . Ref ) ) ;
192+
193+ var replyProbe = CreateTestProbe ( ) ;
194+ ( await producerProbe . ExpectMsgAsync < ShardingProducerController . RequestNext < Job > > ( ) )
195+ . AskNextTo ( new ShardingProducerController . MessageWithConfirmation < Job > ( "entity-1" , new Job ( "msg-1" ) ,
196+ replyProbe . Ref ) ) ;
197+ await replyProbe . ExpectMsgAsync < Done > ( ) ;
198+
199+ ( await producerProbe . ExpectMsgAsync < ShardingProducerController . RequestNext < Job > > ( ) )
200+ . AskNextTo ( new ShardingProducerController . MessageWithConfirmation < Job > ( "entity-2" , new Job ( "msg-2" ) ,
201+ replyProbe . Ref ) ) ;
202+ await replyProbe . ExpectMsgAsync < Done > ( ) ;
203+ }
204+
205+ private class Consumer : ReceiveActor
206+ {
207+ private readonly TestProbe _consumerProbe ;
208+ private readonly IActorRef _consumerController ;
209+ private readonly IActorRef _deliveryAdapter ;
210+
211+ public Consumer ( IActorRef consumerController , TestProbe consumerProbe )
212+ {
213+ _consumerController = consumerController ;
214+ _consumerProbe = consumerProbe ;
215+
216+ var self = Self ;
217+ _deliveryAdapter = Context . ActorOf (
218+ act =>
219+ {
220+ act . Receive < ConsumerController . Delivery < Job > > ( ( delivery , ctx ) =>
221+ {
222+ self . Forward ( new JobDelivery ( delivery . Message , delivery . ConfirmTo , delivery . ProducerId ,
223+ delivery . SeqNr ) ) ;
224+ } ) ;
225+ } , "delivery-adapter" ) ;
226+
227+ Receive < JobDelivery > ( job => { _consumerProbe . Ref . Tell ( job ) ; } ) ;
228+ }
229+
230+ protected override void PreStart ( )
231+ {
232+ _consumerController . Tell ( new ConsumerController . Start < Job > ( _deliveryAdapter ) ) ;
233+ }
234+ }
235+ }
0 commit comments