and what the different tools are meant to be used for
we don't editorialize them much
Is there anyone out there that can help me get my head around the implications that leveraging persistence in an existing actor. I've changed the base class and am moving some "commands" over to the construct events, persist, process event pattern but am getting what "feels" like asynchronous processing of the event processing. Is it true to say that:
1) A message is received
2) Message is treated as a command and an event is created from the command
3) Persist is called supplying the event to persist and a handler to be invoked after successful persistence
4) Next message is pulled from the mailbox and processed
5) Persistence completes
6) Handler supplied in 3 is invoked with persisted event
That's how it feels, so I wonder if that is accurate, and is the fix to complete the moving over from messages to Commands/Events and it'll be okay.
Question around
async methods which return data (to an Ask) and PipeTo: would using ContinueWith on faulted tasks be a reasonable way to 'catch' and handle exceptions thrown. Example: Receive<SomeRequest>(request =>
{
GetSomethingAsync(request)
.ContinueWith(t => new FailedResponse(t.Exception), TaskContinuationOptions.OnlyOnFaulted)
.PipeTo(Sender);
});
part of the issue there though is that you'll never get anything on that
PipeTo if the task succeeds normally I don't think
since
PipeTo is continuing off of the OnlyOnFaulted Task
I have an example of what I usually do here...
// pipe the result of our markets to ourselves
_exchangeFeedClient.HttpClient.Markets.GetProducts(cts.Token).ContinueWith<object>(tr =>
{
if (tr.IsCanceled || tr.IsFaulted)
return new Status.Failure(tr.Exception);
return tr.Result;
}).PipeTo(Self);
I have a catch-all continuation
that I use to handle any case
success or failure
and then my actor who receives this...
Receive<Status.Failure>(s =>
{
_currentRetryPolicy = _currentRetryPolicy.Retry(); // update the retry data
_log.Warning(
"Request for list of exchange trade products timed out from {0}. Retrying in {1} seconds with {2} attempts remaining.",
_exchangeFeedClient.Endpoints.RestEndpointUri, _currentRetryPolicy.CurrentRetryInterval,
_currentRetryPolicy.RemainingAttempts);
AcquireProductsFeed(_currentRetryPolicy.CurrentRetryInterval);
}, failure => _currentRetryPolicy.CanRetry);
Receive<Status.Failure>(s =>
{
_log.Error("ERROR: unable to contact exchange at {0} after {1} attempts. Aborting.",
_exchangeFeed.HttpClient.Endpoints.RestEndpointUri, _retryPolicyProvider.MaxNrOfAttempts);
_currentRetryPolicy.Retry(); // throw a HopelessOperation exception on purpose
}, failure => !_currentRetryPolicy.CanRetry);
// Received data from the exchange API.
Receive<ApiResult<IEnumerable<Product>>>(products =>
{
if (products.Status == HttpStatusCode.OK)
{
_tradeProducts.AddRange(products.Result);
_log.Info("Received products from exchange: {0}", string.Join(",", _tradeProducts.Select(x => x.id)));
BecomePublishing();
}
else if (_currentRetryPolicy.CanRetry
) // something went wrong with our web request, but we have retry attempts remaining
{
_currentRetryPolicy = _currentRetryPolicy.Retry(); // update the retry data
_log.Warning(
"Received error with our HTTP request ({0}): {1}. Retrying in {2} with {3} attempts left.",
products.Status, products.Message,
_currentRetryPolicy.CurrentRetryInterval, _currentRetryPolicy.RemainingAttempts);
AcquireProductsFeed(_currentRetryPolicy.CurrentRetryInterval);
}
else // web request failed, no retry attempts remaining
{
_log.Error(
"ERROR: unable to contact exchange at {0} after {1} attempts. HTTP Status: {2}, Message: {3}. Aborting.",
_exchangeFeed.HttpClient.Endpoints.RestEndpointUri,
_retryPolicyProvider.MaxNrOfAttempts, products.Status, products.Message);
_currentRetryPolicy.Retry(); // throw a HopelessOperation exception on purpose
}
});AcquireProductsFeed(_currentRetryPolicy.CurrentRetryInterval); causes me to re-run the task that failed
the implementation I'm using here is an exponential-backoff type approach
I wrote that stuff by hand but you can use a library like Polly to get that for free
@goodisontoffee if you call the normal
Persist method
the actor will complete its persistence operation before the next message from the mailbox can be processed
if you call
PersistAsync
it will behave as you've described
because the journal itself is implemented as an actor
and it will receive the messages in its own mailbox in that order
So Persist blocks new message consumption until the persistence is complete.
the actor is just stashing messages until it gets an acknowledgement that the persist operation has completed
and then it invokes the callback
so it's not literally blocking a thread
Hi all, trying to build client to remote akka system using association. Using test:
public class AssociationErrorTest
{
public class EchoActor : ReceiveActor
{
public EchoActor(){Receive<object>(m => Sender.Tell(m));}
}
[Fact]
public async Task Test_association()
{
var server = ActorSystem.Create("server", @"akka { actor.provider = ""Akka.Remote.RemoteActorRefProvider, Akka.Remote""
remote.dot-netty.tcp { port = 10001
hostname = localhost}}");
var serverActor = server.ActorOf<EchoActor>(nameof(EchoActor));
var client = ActorSystem.Create("client", @"akka{ actor.provider = ""Akka.Remote.RemoteActorRefProvider, Akka.Remote""
remote.dot-netty.tcp { port = 0
hostname = localhost}}");
var remoteActorRef = await client.ActorSelection(@"akka.tcp://server@localhost:10001/user/EchoActor").ResolveOne(TimeSpan.FromSeconds(5));
await remoteActorRef.Ask<string>("Hello");
}
}if I run it as single test, all is OK . But if I run it with bunch of test, actor selection hangs. Got error from internals:
17-10-05 17:50:08.366 [DBG TH193] Src:[remoting
"Associated [akka.tcp://server@localhost:10001] <- akka.tcp://client@localhost:5703"
17-10-05 17:50:08.375 [DBG TH201] Src:[endpointWriter#116268940]
"Associated [akka.tcp://client@localhost:5703] -> akka.tcp://server@localhost:10001"
17-10-05 17:50:08.375 [DBG TH201] Src:[endpointWriter#116268940]
Drained buffer with maxWriteCount: 50, fullBackoffCount: 1,smallBackoffCount: 0, noBackoffCount: 0,adaptiveBackoff: 1000
17-10-05 17:50:08.388 [DBG TH201] Src:[client)
Resolve of path sequence [/"temp/N"] failedlooks like client system cannot locate temp actor during Ask()
Can anybody suggest reasons? May be I need some additional configuration ?
issue is either going to be a static variable that is shared among the actor system instances and this test is running with AppDomains turned off
or one of the remote actor systems is replying to an actor who has since terminated
may be something inside akka system itself?
the name counter we use for naming temporary actors
might be one
you are right
I see letter for temp actor always increasing ))
ok, will run such tests in separate domains
I'm having trouble doing more than one transition in an FSM. Not sure if it's a problem with the FSM code or if it's a problem with the unit tests. Simple and full reproducible example at https://stackoverflow.com/questions/46588702/why-does-this-fsm-only-succeed-at-one-state-transition if anyone wants to take a look.