I suspect
^^^
I didnt have a reference to it at all
thanks again for the help
one question, what triggers a rebalance?
i am having an issue using UnboundedStash
below is a cut-down actor implementation that i am using
public class ClassName : ReceiveActor, IWithUnboundedStash
{
public IStash Stash { get; set; }
public ClassName()
{
ReceiveAny(_ => Stash.Stash());
}
void BecomeLoaded()
{
Stash.UnstashAll();
}
}
the problem i am having is that when BecomeLoaded() is called, i am getting a null pointer exception on Stash
i have not tried sending a message to it to see if one is also thrown on ReceiveAny
any thoughts on this? i can obviously do a null check before calling UnstashAll, but concerned that its not actually working
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.