Upgrading to 1.3.5 from 1.2.3, and I'm starting to see some random serialization exceptions upon restarting some nodes:
Cannot find serializer with id [7]. The most probable reason is that the configuration entry 'akka.actor.serializers' is not in sync between the two systems.
which is weird because that's Akka's persistence serializer, which is loaded automatically
@ismaelhamed have you specified persistence config as fallback explicitly? If I remember correctly, after 1.3 there were some problems with loading configurations - because we started supporting .net standard - i.e. app.config were no longer used due to lack of ConfigurationManager API.
@Greatsamps stash can be null when reached from constructor, it's set automatically after constructor call.
thanks for your help
or thinking about this, becasue the constructer has not completed, would that put the stash functionality on hold? should i refactor this?
I am trying to use the amqp stream in combination with RabbitMQ. I intend to open source the example, but I am still figuring out how the do communication with clients using rabbitMQ streams. I have the following code in the PortalCoordinator (which is going to act as the server to clien portal).
public class AmqpPortalCoordinator : ReceiveActor
{
private readonly IActorRef chatserver;
private readonly ActorMaterializer mat;
const string inboundQueue = "InboundWeChatServer";
private readonly AmqpConnectionDetails connectionSettings;
private Task incomingTask;
public AmqpPortalCoordinator(IActorRef chatserver)
{
this.chatserver = chatserver;
mat = ActorMaterializer.Create(Context);
connectionSettings = AmqpConnectionDetails.Create("localhost", 5672)
.WithAutomaticRecoveryEnabled(true)
.WithNetworkRecoveryInterval(TimeSpan.FromSeconds(1));
var queueDeclaration = QueueDeclaration
.Create(inboundQueue)
.WithDurable(true)
.WithAutoDelete(false);
var queueSource = NamedQueueSourceSettings.Create(connectionSettings, inboundQueue).WithDeclarations(queueDeclaration);
var amqpSource = AmqpSource.CommittableSource(queueSource, 1000);
var messageSource = amqpSource.Select(m =>
{
var bytes = m.Message.Bytes;
m.Ack();
return MessagePackSerializer.Typeless.Deserialize(bytes.ToArray());
});
incomingTask = messageSource.RunForeach(msg => chatserver.Tell(msg), mat);
}
}
Or am I missing the point using amqp sources?
My biggest concern is actually how to dispose the source, when the server stops for some reason
public class AmqpPortalCoordinator : ReceiveActor, IDisposable
{
private readonly IActorRef chatserver;
private readonly ActorMaterializer mat;
const string inboundQueue = "InboundWeChatServer";
private readonly AmqpConnectionDetails connectionSettings;
private UniqueKillSwitch killswitch;
private Task sourceTask;
public AmqpPortalCoordinator(IActorRef chatserver)
{
this.chatserver = chatserver;
mat = ActorMaterializer.Create(Context);
connectionSettings = AmqpConnectionDetails.Create("localhost", 5672)
.WithAutomaticRecoveryEnabled(true)
.WithNetworkRecoveryInterval(TimeSpan.FromSeconds(1));
var queueDeclaration = QueueDeclaration
.Create(inboundQueue)
.WithDurable(true)
.WithAutoDelete(false);
var queueSource = NamedQueueSourceSettings.Create(connectionSettings, inboundQueue).WithDeclarations(queueDeclaration);
var amqpSource = AmqpSource.CommittableSource(queueSource, 50);
var (killswitch, task) = amqpSource
.ViaMaterialized(KillSwitches.Single<CommittableIncomingMessage>(), Keep.Right)
.ToMaterialized(Sink.ForEach<CommittableIncomingMessage>((m) => {
var bytes = m.Message.Bytes;
m.Ack();
chatserver.Tell(MessagePackSerializer.Typeless.Deserialize(bytes.ToArray()));
}), Keep.Both).Run(mat);
this.killswitch = killswitch;
this.sourceTask = task;
}
public void Dispose()
{
killswitch.Shutdown();
sourceTask.Wait();
}
}
public class RequestMessage
{
public class Response
{}
}public class PutwallActorTests : TestKit
{
[Fact]
public void RequestUpdateStateSentMultipleTimes_OnlyOneResponseBack()
{
var probe = CreateTestProbe();
IActorRef putwallActor = Sys.ActorOf(PutwallActor.Props());
var guidId = Guid.NewGuid();
for (int i = 0; i < 100; i++)
{
putwallActor.Tell(new PutwallActor.CubbyMessage.RequestUpdateState(true, guidId), probe.Ref);
}
var cubbyStateResponse = probe.ExpectMsg<PutwallActor.CubbyMessage.RequestUpdateState.Response>();
Assert.Equal(guidId, cubbyStateResponse.MessageId);
}
}
I know this ^^^ isn't right, but I'm not sure where to go from here.
For example if I added two response message statements like:
var cubbyStateResponse = probe.ExpectMsg<PutwallActor.CubbyMessage.RequestUpdateState.Response>();
var cubbyStateResponse2 = probe.ExpectMsg<PutwallActor.CubbyMessage.RequestUpdateState.Response>();The test would fail, but my functionality would be correct because the unit test times out on the second statement.
Thanks!
of course if your actor replies 5 minutes later with a message, the test won't catch it, but hopefully that doesn't occur :)
[Fact]
public void RequestUpdateStateSentMultipleTimes_OnlyOneResponseBack()
{
var probe = CreateTestProbe();
IActorRef putwallActor = Sys.ActorOf(PutwallActor.Props());
var guidId = Guid.NewGuid();
putwallActor.Tell(new PutwallActor.CubbyMessage.RequestUpdateState(true, guidId), probe.Ref);
var cubbyStateResponse = probe.ExpectMsg<PutwallActor.CubbyMessage.RequestUpdateState.Response>();
Assert.Equal(guidId, cubbyStateResponse.MessageId);
for (int i = 0; i < 200; i++)
{
putwallActor.Tell(new PutwallActor.CubbyMessage.RequestUpdateState(true, guidId), probe.Ref);
probe.ExpectNoMsg(TimeSpan.FromMilliseconds(1));
}
}
Hello everyone. Looking for some pointers on the right way to do the following with Akka:
Actor A needs to send a message to Actor B, however, they can’t see each other, and the only communication channel between these two actors is an Azure Blobstore that they both can read/write.
Should I be looking at implementing a custom mailbox? Custom transport? Or something else? Any recommendations welcome.
and these actors are running on different processes I take it?
I could easily serialize my some messages and make it work, but I’d prefer to use the Akka infrastructure
I basically want that, but replacing the channel with something else that will take care of persisting the message to/from Azure Blobstore (or whatever the shared place is)
