Handling Events from a Service running in a Task in an AKKA.NET actor

Question

I'm using the Prism framework for my WPF application. I have a producer service that runs in a Task and raises a CompositePresentationEvent when a file is found. I have an Akka actor that subscribes to the event. The actor's handler looks is simple, it looks like this:

private void HandleFileReceive(FileEventArgs args)
{
    Self.Tell(new FileReceived(args.File));
}

When the event hits the handler above, I receive a System.NotSupportedException with this message: "There is no active ActorContext, this is most likely due to use of async operations from within this actor".

I assume this is because the service is running in a different thread than the actor's handler. Is there a way to handle this type of thing in Akka.NET?

I'm not completely opposed to writing new Actors that do the job of the Services that are needed for my situations. The issue is that, depending on some settings in a file, the service will be different. Currently, I'm handling this using MEF and getting the correct implementer of my given interface from the IoC container. I'd like to continue to keep the concrete implementations of the producers abstracted from the core code (where the actors are).

Any suggestions on getting passed this (perceived) threading issue and/or dynamically generating a ProducerActor that implements a given interface?

Thanks

-g

Answer 1

I ended up creating an actor to handle the response retrieval and changing the producer interface so the concrete producers only need to implement a single, simple method. The old producers were in charge of waiting a config-driven amount of time before checking for responses, but the new code utilizes AKKA's scheduler to tell the retriever, at the same config-driven interval, when to check for responses. I've provided a sample below, but my code is a little more complex with error handling and such.

In the code below, you see the interface for IProducer and the concrete implementations: LocalProducer and EmailProducer. In the real code, these have attributes that tell the container some other information that is used to get the correct implementation from the container. The ConsumerActor is the parent actor for this scenario and handles the Consume message. It creates a ResponseRetrieverActor in its constructor and schedules a repeated tell for the ResponseRetrieverActor so that the retriever checks for responses at a given interval.

In the ResponseRetrieverActor's RetrieveResponses handler is where the IoC magic happens. I get the transport type from a factory method (not shown here -- it is actually set in a config file of sorts). The transport type is used to retrieve the correct producer from the IoC container using the aforementioned attributes on the concrete Producers. Finally, the producer is used to get the responses and tells the parent, which is the consumer, about each file in the list.

public interface IProducer
{
   List<string> GetResponses();
}

[Export(typeof(IProducer))] // Other attributes needed for MEF Export to differentiate the multiple IProducer implementations
public LocalProducer : IProducer
{
   public List<string> GetResponse()
   {
       // get files from a directory
   }
}

[Export(typeof(IProducer))]
public EmailProducer : IProducer
{
   public List<string> GetResponse()
   {
       // get files from email account
   }
}

public class ConsumerActor
{
    public class Consume
    {
       public Consume(string file) { this.File = file; }

       public string File { get; set; }
    }

    public ConsumerActor() 
    {
       _retriever = Context.ActorOf(Props.Create<ResponseRetrieverActor>(), "retriever");

       var interval = 10000;
       Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(0, interval, _retriever, new ResponseRetrieverActor.RetrieveResponses(), Self);
       Start(); 
    }

    private void Start()
    {
        Receive<Consume>(msg => 
        {
           // do something with the msg.File 
        });
    }

    private IActorRef _retriever;
}
public class ResponseRetrieverActor
{
  public class RetrieveResponses { }

  public ConsumerActor() { Start(); }

  private void Start()
  {
    Receive<RetrieveResponses>(msg => HandleRetrieveResponses());  
  }

  private void HandleRetrieveResponses()
  {
     var transportType = TransportFactory.GetTransportType(); // Gets transport protocol for the producer we need to use (Email, File, ect.)
     var producer = ServiceLocator.Current.GetInstance<IProducer>(transportType); // Gets a producer from the IoC container for the specified transportType

     var responses = producer.GetResponses();

     foreach(var response in responseFiles)
     {
         Context.Parent.Tell(new ConsumerActor.Consume(response));
     }
  }
}