Public API cleanup

ProtoActor.sln

@@ -203,7 +203,7 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Server", "examples\remotech
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Client", "examples\remotechannels\Client\Client.csproj", "{5EACFB26-C757-452C-A536-E633BC352A69}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Messages", "examples\remotechannels\Messages\Messages.csproj", "{D33722CB-96B8-4B9C-AB59-B18B7FF28539}"
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Common", "examples\remotechannels\Common\Common.csproj", "{D33722CB-96B8-4B9C-AB59-B18B7FF28539}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Proto.Cluster.AmazonECS", "src\Proto.Cluster.AmazonECS\Proto.Cluster.AmazonECS.csproj", "{F5EB4AD3-5BCB-48E9-B974-A93EB19CE43C}"
 EndProject

examples/remotechannels/Client/Client.csproj

@@ -8,7 +8,7 @@
 
  <ItemGroup>
  <ProjectReference Include="..\..\..\src\Proto.Remote\Proto.Remote.csproj" />
- <ProjectReference Include="..\Messages\Messages.csproj" />
+ <ProjectReference Include="..\Common\Common.csproj" />
  </ItemGroup>
 
 </Project>

examples/remotechannels/Client/Program.cs

@@ -1,7 +1,6 @@
 using System;
-using Messages;
+using Common;
 using Proto;
-using Proto.Channels;
 using Proto.Remote;
 using Proto.Remote.GrpcNet;
 using static System.Threading.Channels.Channel;

src/Proto.Actor/Channels/ChannelPublisherActor.cs → examples/remotechannels/Common/ChannelPublisherActor.cs

@@ -6,11 +6,10 @@
 using System.Collections.Generic;
 using System.Threading.Channels;
 using System.Threading.Tasks;
-using JetBrains.Annotations;
+using Proto;
 
-namespace Proto.Channels;
+namespace Common;
 
-[PublicAPI]
 public static class ChannelPublisher
 {
  /// <summary>
@@ -40,7 +39,6 @@ public static PID StartNew<T>(IRootContext context, Channel<T> channel, string n
  }
 }
 
-[PublicAPI]
 public class ChannelPublisherActor<T> : IActor
 {
  private readonly HashSet<PID> _subscribers = new();
@@ -56,13 +54,17 @@ public Task ReceiveAsync(IContext context)
  }
 
  break;
+
  case PID subscriber:
  _subscribers.Add(subscriber);
  context.Watch(subscriber);
+ context.Respond(new Subscribed());
  break;
+
  case Terminated terminated:
  _subscribers.Remove(terminated.Who);
  break;
+
  case T typed:
  foreach (var sub in _subscribers)
  {

src/Proto.Actor/Channels/ChannelSubscriberActor.cs → examples/remotechannels/Common/ChannelSubscriberActor.cs

@@ -5,11 +5,10 @@
 // -----------------------------------------------------------------------
 using System.Threading.Channels;
 using System.Threading.Tasks;
-using JetBrains.Annotations;
+using Proto;
 
-namespace Proto.Channels;
+namespace Common;
 
-[PublicAPI]
 public static class ChannelSubscriber
 {
  /// <summary>
@@ -21,24 +20,29 @@ public static class ChannelSubscriber
  /// <param name="channel">The channel to write messages to</param>
  /// <typeparam name="T">The Type of channel elements</typeparam>
  /// <returns></returns>
- public static PID StartNew<T>(IRootContext context, PID publisher, Channel<T> channel)
+ public static async Task<PID> StartNew<T>(IRootContext context, PID publisher, Channel<T> channel)
  {
- var props = Props.FromProducer(() => new ChannelSubscriberActor<T>(publisher, channel));
+ var tcs = new TaskCompletionSource();
+ var props = Props.FromProducer(() => new ChannelSubscriberActor<T>(publisher, channel, tcs));
  var pid = context.Spawn(props);
+
+ await tcs.Task;
  return pid;
  }
 }
 
-[PublicAPI]
 public class ChannelSubscriberActor<T> : IActor
 {
  private readonly Channel<T> _channel;
+ private readonly TaskCompletionSource _subscribed;
  private readonly PID _publisher;
 
- public ChannelSubscriberActor(PID publisher, Channel<T> channel)
+ public ChannelSubscriberActor(PID publisher, Channel<T> channel, TaskCompletionSource subscribed)
+
  {
  _publisher = publisher;
  _channel = channel;
+ _subscribed = subscribed;
  }
 
  public async Task ReceiveAsync(IContext context)
@@ -49,15 +53,24 @@ public async Task ReceiveAsync(IContext context)
  context.Watch(_publisher);
  context.Request(_publisher, context.Self);
  break;
+
+ case Subscribed:
+ _subscribed.SetResult();
+ break;
+
  case Stopping:
  _channel.Writer.Complete();
  break;
+
  case Terminated t when t.Who.Equals(_publisher):
  _channel.Writer.Complete();
  break;
+
  case T typed:
  await _channel.Writer.WriteAsync(typed);
  break;
  }
  }
-}
+}
+
+public record Subscribed;

examples/remotechannels/Messages/Messages.csproj → examples/remotechannels/Common/Common.csproj

@@ -5,4 +5,8 @@
  <LangVersion>10</LangVersion>
  </PropertyGroup>
 
+ <ItemGroup>
+ <ProjectReference Include="..\..\..\src\Proto.Actor\Proto.Actor.csproj" />
+ </ItemGroup>
+
 </Project>

examples/remotechannels/Common/Messages.cs

@@ -0,0 +1,3 @@
+namespace Common;
+
+public record MyMessage(int Value){}

examples/remotechannels/Server/Program.cs

@@ -1,8 +1,7 @@
 using System;
 using System.Threading.Tasks;
-using Messages;
+using Common;
 using Proto;
-using Proto.Channels;
 using Proto.Remote;
 using Proto.Remote.GrpcNet;
 using static Proto.Remote.GrpcNet.GrpcNetRemoteConfig;

examples/remotechannels/Server/Server.csproj

@@ -8,7 +8,7 @@
 
  <ItemGroup>
  <ProjectReference Include="..\..\..\src\Proto.Remote\Proto.Remote.csproj" />
- <ProjectReference Include="..\Messages\Messages.csproj" />
+ <ProjectReference Include="..\Common\Common.csproj" />
  </ItemGroup>
 
 </Project>

src/Proto.Actor/ActorSystem.cs

@@ -78,7 +78,12 @@ public ActorSystem(ActorSystemConfig config)
  /// Allows to access the stop cancellation token and stop reason.
  /// Use <see cref="ShutdownAsync"/> to stop the actor system.
  /// </summary>
- public Stopper Stopper { get; }
+ internal Stopper Stopper { get; }
+
+ /// <summary>
+ /// For stopped <see cref="ActorSystem"/>, returns the reason for the shutdown.
+ /// </summary>
+ public string StoppedReason => Stopper.StoppedReason;
 
  /// <summary>
  /// Manages all the guardians in the actor system.
@@ -117,13 +122,13 @@ public ActorSystem(ActorSystemConfig config)
 
  private void RunThreadPoolStats()
  {
- var metricTags = new KeyValuePair<string, object?>[]{ new("id", Id), new("address", Address)};
+ var metricTags = new KeyValuePair<string, object?>[] {new("id", Id), new("address", Address)};
 
  var logger = Log.CreateLogger(nameof(ThreadPoolStats));
  _ = ThreadPoolStats.Run(TimeSpan.FromSeconds(5),
  t => {
  //collect the latency metrics
- if(Metrics.Enabled)
+ if (Metrics.Enabled)
  ActorMetrics.ThreadPoolLatency.Record(t.TotalSeconds, metricTags);
 
  //does it take longer than 1 sec for a task to start executing?
@@ -144,7 +149,7 @@ private void RunThreadPoolStats()
  /// </summary>
  /// <param name="reason">Shutdown reason</param>
  /// <returns></returns>
- public Task ShutdownAsync(string reason="")
+ public Task ShutdownAsync(string reason = "")
  {
  try
  {
@@ -195,15 +200,15 @@ public RootContext NewRoot(MessageHeader? headers = null, params Func<Sender, Se
  /// </summary>
  /// <param name="props"></param>
  /// <returns></returns>
- public Props ConfigureProps(Props props) => Config.ConfigureProps(props);
- 
+ internal Props ConfigureProps(Props props) => Config.ConfigureProps(props);
+
  /// <summary>
  /// Applies props configuration delegate for system actors.
  /// </summary>
  /// <param name="name"></param>
  /// <param name="props"></param>
  /// <returns></returns>
- public Props ConfigureSystemProps(string name, Props props) => Config.ConfigureSystemProps(name, props);
+ internal Props ConfigureSystemProps(string name, Props props) => Config.ConfigureSystemProps(name, props);
 
  /// <summary>
  /// Stops the actor system with reason = "Disposed"

src/Proto.Actor/Context/ActorContext.cs

@@ -71,11 +71,6 @@ public ActorContext(ActorSystem system, Props props, PID? parent, PID self, IMai
 
  public TimeSpan ReceiveTimeout { get; private set; }
 
- public void Stash()
- {
- if (_messageOrEnvelope is not null) EnsureExtras().Stash.Push(_messageOrEnvelope);
- }
-
  public void Respond(object message)
  {
  if (Sender is not null)
@@ -714,19 +709,6 @@ private async ValueTask RestartAsync()
  Self.SendSystemMessage(System, ResumeMailbox.Instance);
 
  await InvokeUserMessageAsync(Started.Instance);
-
- if (_extras?.Stash is not null)
- {
- var currentStash = new Stack<object>(_extras.Stash);
- _extras.Stash.Clear();
-
- //TODO: what happens if we hit a failure here?
- while (currentStash.Any())
- {
- var msg = currentStash.Pop();
- await InvokeUserMessageAsync(msg);
- }
- }
  }
 
  private ValueTask DisposeActorIfDisposable()

src/Proto.Actor/Context/ActorContextDecorator.cs

@@ -48,9 +48,6 @@ public virtual Task Receive(MessageEnvelope envelope) =>
  public virtual void Respond(object message) =>
  _context.Respond(message);
 
- public virtual void Stash() =>
- _context.Stash();
-
  public virtual PID SpawnNamed(Props props, string name, Action<IContext>? callback=null) =>
  _context.SpawnNamed(props, name, callback);
 

src/Proto.Actor/Context/ActorContextExtras.cs

@@ -4,7 +4,6 @@
 // </copyright>
 // -----------------------------------------------------------------------
 using System;
-using System.Collections.Generic;
 using System.Collections.Immutable;
 using System.Threading;
 using Proto.Utils;
@@ -25,12 +24,11 @@ public sealed class ActorContextExtras: IDisposable
  public ImmutableHashSet<PID> Children { get; private set; } = ImmutableHashSet<PID>.Empty;
  public Timer? ReceiveTimeoutTimer { get; private set; }
  public RestartStatistics RestartStatistics { get; } = new(0, null);
- public Stack<object> Stash { get; } = new();
  public ImmutableHashSet<PID> Watchers { get; private set; } = ImmutableHashSet<PID>.Empty;
  public IContext Context { get; }
  public CancellationTokenSource CancellationTokenSource { get; } = new();
 
- public TypeDictionary<object, ActorContextExtras> Store { get; } = new(5, 1);
+ internal TypeDictionary<object, ActorContextExtras> Store { get; } = new(5, 1);
 
  public void InitReceiveTimeoutTimer(Timer timer) => ReceiveTimeoutTimer = timer;
 

src/Proto.Actor/Context/IContext.cs

@@ -44,11 +44,6 @@ public interface IContext : ISenderContext, IReceiverContext, ISpawnerContext, I
  /// <param name="header"></param>
  void Respond(object message, MessageHeader header) => Respond(new MessageEnvelope(message, null, header));
 
- /// <summary>
- /// Stashes the current message on a stack for re-processing when the actor restarts.
- /// </summary>
- void Stash();
-
  /// <summary>
  /// Registers the actor as a watcher for the specified PID. When the PID terminates the watcher is notified with <see cref="Terminated"/> message.
  /// </summary>

src/Proto.Actor/Context/ISpawnerContext.cs

@@ -26,6 +26,7 @@ public static class SpawnerContextExtensions
  /// <summary>
  /// Spawns a new child actor based on props and named with a unique ID.
  /// </summary>
+ /// <param name="self"></param>
  /// <param name="props">The Props used to spawn the actor</param>
  /// <returns>The PID of the child actor</returns>
  public static PID Spawn(this ISpawnerContext self, Props props)
@@ -36,6 +37,7 @@ public static PID Spawn(this ISpawnerContext self, Props props)
  /// <summary>
  /// Spawns a new child actor based on props and named with a unique ID.
  /// </summary>
+ /// <param name="self"></param>
  /// <param name="props">The Props used to spawn the actor</param>
  /// <param name="callback"></param>
  /// <returns>The PID of the child actor</returns>

src/Proto.Actor/Deduplication/DeduplicationContext.cs

@@ -16,6 +16,7 @@ namespace Proto.Deduplication;
 /// Extracts the deduplication key from the message.
 /// </summary>
 /// <typeparam name="T">Type of the key</typeparam>
+/// <param name="envelope">Message to extract from</param>
 /// <param name="key">The key should be returned in this variable</param>
 /// <returns>Returns true if the key was successfully extracted, false otherwise</returns>
 public delegate bool TryGetDeduplicationKey<T>(MessageEnvelope envelope, out T? key);

src/Proto.Actor/Messages/MessageEnvelope.cs

@@ -114,7 +114,7 @@ public MessageEnvelope WithHeader(string key, string value)
  /// <summary>
  /// Extends the message envelope with additional headers.
  /// </summary>
- /// <param name="header"></param>
+ /// <param name="items"></param>
  /// <returns>New envelope</returns>
  public MessageEnvelope WithHeaders(IEnumerable<KeyValuePair<string, string>> items)
  {

src/Proto.Actor/PID.cs

@@ -61,7 +61,7 @@ internal void SendUserMessage(ActorSystem system, object message)
  reff.SendUserMessage(this, message);
  }
 
- public void SendSystemMessage(ActorSystem system, SystemMessage sys)
+ internal void SendSystemMessage(ActorSystem system, SystemMessage sys)
  {
  var reff = Ref(system) ?? system.ProcessRegistry.Get(this);
  reff.SendSystemMessage(this, sys);

src/Proto.Actor/Props/Props.cs

@@ -60,16 +60,16 @@ public sealed record Props
  public ImmutableList<Func<Sender, Sender>> SenderMiddleware { get; init; } =
  ImmutableList<Func<Sender, Sender>>.Empty;
 
- public Receiver? ReceiverMiddlewareChain { get; init; }
- public Sender? SenderMiddlewareChain { get; init; }
+ internal Receiver? ReceiverMiddlewareChain { get; init; }
+ internal Sender? SenderMiddlewareChain { get; init; }
 
  /// <summary>
  /// List of decorators for the actor context
  /// </summary>
  public ImmutableList<Func<IContext, IContext>> ContextDecorator { get; init; } =
  ImmutableList<Func<IContext, IContext>>.Empty;
 
- public Func<IContext, IContext>? ContextDecoratorChain { get; init; }
+ internal Func<IContext, IContext>? ContextDecoratorChain { get; init; }
 
  /// <summary>
  /// Delegate that creates the actor and wires it with context and mailbox.