Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[api-docs] Added TBD placeholders for Akka.Streams.csproj #2448

Merged
merged 1 commit into from
Jan 13, 2017
Merged

[api-docs] Added TBD placeholders for Akka.Streams.csproj #2448

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
235 changes: 233 additions & 2 deletions src/core/Akka.Streams/ActorMaterializer.cs
View file
Open in desktop

Large diffs are not rendered by default.

176 changes: 176 additions & 0 deletions src/core/Akka.Streams/Actors/ActorPublisher.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,28 +19,63 @@ namespace Akka.Streams.Actors
{
#region Internal messages

/// <summary>
/// TBD
/// </summary>
/// <typeparam name="T">TBD</typeparam>
public sealed class Subscribe<T> : INoSerializationVerificationNeeded, IDeadLetterSuppression
{
/// <summary>
/// TBD
/// </summary>
public readonly ISubscriber<T> Subscriber;
/// <summary>
/// TBD
/// </summary>
/// <param name="subscriber">TBD</param>
public Subscribe(ISubscriber<T> subscriber)
{
Subscriber = subscriber;
}
}

/// <summary>
/// TBD
/// </summary>
[Serializable]
public enum LifecycleState
{
/// <summary>
/// TBD
/// </summary>
PreSubscriber,
/// <summary>
/// TBD
/// </summary>
Active,
/// <summary>
/// TBD
/// </summary>
Canceled,
/// <summary>
/// TBD
/// </summary>
Completed,
/// <summary>
/// TBD
/// </summary>
CompleteThenStop,
/// <summary>
/// TBD
/// </summary>
ErrorEmitted
}

#endregion

/// <summary>
/// TBD
/// </summary>
public interface IActorPublisherMessage: IDeadLetterSuppression { }

/// <summary>
Expand All @@ -50,8 +85,15 @@ public interface IActorPublisherMessage: IDeadLetterSuppression { }
[Serializable]
public sealed class Request : IActorPublisherMessage
{
/// <summary>
/// TBD
/// </summary>
public readonly long Count;

/// <summary>
/// TBD
/// </summary>
/// <param name="count">TBD</param>
public Request(long count)
{
Count = count;
Expand All @@ -75,6 +117,9 @@ public Request(long count)
[Serializable]
public sealed class Cancel : IActorPublisherMessage
{
/// <summary>
/// TBD
/// </summary>
public static readonly Cancel Instance = new Cancel();
private Cancel() { }
}
Expand All @@ -87,6 +132,9 @@ private Cancel() { }
[Serializable]
public sealed class SubscriptionTimeoutExceeded : IActorPublisherMessage
{
/// <summary>
/// TBD
/// </summary>
public static readonly SubscriptionTimeoutExceeded Instance = new SubscriptionTimeoutExceeded();
private SubscriptionTimeoutExceeded() { }
}
Expand Down Expand Up @@ -144,6 +192,7 @@ private SubscriptionTimeoutExceeded() { }
/// failure, completed or canceled.
/// </para>
/// </summary>
/// <typeparam name="T">TBD</typeparam>
public abstract class ActorPublisher<T> : ActorBase
{
private readonly ActorPublisherState _state = ActorPublisherState.Instance.Apply(Context.System);
Expand All @@ -155,6 +204,9 @@ public abstract class ActorPublisher<T> : ActorBase
// case and stop fields are used only when combined with LifecycleState.ErrorEmitted
private OnErrorBlock _onError;

/// <summary>
/// TBD
/// </summary>
protected ActorPublisher()
{
SubscriptionTimeout = Timeout.InfiniteTimeSpan;
Expand Down Expand Up @@ -227,6 +279,8 @@ public bool IsActive
/// <see cref="IsActive"/> and <see cref="TotalDemand"/> &gt; 0,
/// otherwise <see cref="OnNext"/> will throw <see cref="IllegalStateException"/>.
/// </summary>
/// <param name="element">TBD</param>
/// <exception cref="IllegalStateException">TBD</exception>
public void OnNext(T element)
{
switch (_lifecycleState)
Expand Down Expand Up @@ -257,6 +311,7 @@ public void OnNext(T element)
/// Complete the stream. After that you are not allowed to
/// call <see cref="OnNext"/>, <see cref="OnError"/> and <see cref="OnComplete"/>.
/// </summary>
/// <exception cref="IllegalStateException">TBD</exception>
public void OnComplete()
{
switch (_lifecycleState)
Expand Down Expand Up @@ -326,6 +381,8 @@ public void OnCompleteThenStop()
/// Terminate the stream with failure. After that you are not allowed to
/// call <see cref="OnNext"/>, <see cref="OnError"/> and <see cref="OnComplete"/>.
/// </summary>
/// <param name="cause">TBD</param>
/// <exception cref="IllegalStateException">TBD</exception>
public void OnError(Exception cause)
{
switch (_lifecycleState)
Expand Down Expand Up @@ -366,6 +423,7 @@ public void OnError(Exception cause)
/// will be delayed until such <see cref="ISubscriber{T}"/> arrives.
/// </para>
/// </summary>
/// <param name="cause">TBD</param>
public void OnErrorThenStop(Exception cause)
{
switch (_lifecycleState)
Expand Down Expand Up @@ -394,6 +452,12 @@ public void OnErrorThenStop(Exception cause)

#region Internal API

/// <summary>
/// TBD
/// </summary>
/// <param name="receive">TBD</param>
/// <param name="message">TBD</param>
/// <returns>TBD</returns>
protected override bool AroundReceive(Receive receive, object message)
{
if (message is Request)
Expand Down Expand Up @@ -485,6 +549,9 @@ private void CancelSelf()
_demand = 0L;
}

/// <summary>
/// TBD
/// </summary>
public override void AroundPreStart()
{
base.AroundPreStart();
Expand All @@ -496,13 +563,23 @@ public override void AroundPreStart()
}
}

/// <summary>
/// TBD
/// </summary>
/// <param name="cause">TBD</param>
/// <param name="message">TBD</param>
public override void AroundPreRestart(Exception cause, object message)
{
// some state must survive restart
_state.Set(Self, new ActorPublisherState.State(UntypedSubscriber.FromTyped(_subscriber), _demand, _lifecycleState));
base.AroundPreRestart(cause, message);
}

/// <summary>
/// TBD
/// </summary>
/// <param name="cause">TBD</param>
/// <param name="message">TBD</param>
public override void AroundPostRestart(Exception cause, object message)
{
var s = _state.Remove(Self);
Expand All @@ -516,6 +593,9 @@ public override void AroundPostRestart(Exception cause, object message)
base.AroundPostRestart(cause, message);
}

/// <summary>
/// TBD
/// </summary>
public override void AroundPostStop()
{
_state.Remove(Self);
Expand All @@ -533,63 +613,136 @@ public override void AroundPostStop()
#endregion
}

/// <summary>
/// TBD
/// </summary>
public static class ActorPublisher
{
/// <summary>
/// TBD
/// </summary>
/// <typeparam name="T">TBD</typeparam>
/// <param name="ref">TBD</param>
/// <returns>TBD</returns>
public static IPublisher<T> Create<T>(IActorRef @ref) => new ActorPublisherImpl<T>(@ref);
}

/// <summary>
/// TBD
/// </summary>
/// <typeparam name="T">TBD</typeparam>
public sealed class ActorPublisherImpl<T> : IPublisher<T>
{
private readonly IActorRef _ref;

/// <summary>
/// TBD
/// </summary>
/// <param name="ref">TBD</param>
/// <exception cref="ArgumentNullException">TBD</exception>
public ActorPublisherImpl(IActorRef @ref)
{
if(@ref == null) throw new ArgumentNullException(nameof(@ref), "ActorPublisherImpl requires IActorRef to be defined");
_ref = @ref;
}

/// <summary>
/// TBD
/// </summary>
/// <param name="subscriber">TBD</param>
/// <exception cref="ArgumentNullException">TBD</exception>
public void Subscribe(ISubscriber<T> subscriber)
{
if (subscriber == null) throw new ArgumentNullException(nameof(subscriber), "Subscriber must not be null");
_ref.Tell(new Subscribe<T>(subscriber));
}
}

/// <summary>
/// TBD
/// </summary>
public sealed class ActorPublisherSubscription : ISubscription
{
private readonly IActorRef _ref;

/// <summary>
/// TBD
/// </summary>
/// <param name="ref">TBD</param>
/// <exception cref="ArgumentNullException">TBD</exception>
public ActorPublisherSubscription(IActorRef @ref)
{
if (@ref == null) throw new ArgumentNullException(nameof(@ref), "ActorPublisherSubscription requires IActorRef to be defined");
_ref = @ref;
}

/// <summary>
/// TBD
/// </summary>
/// <param name="n">TBD</param>
public void Request(long n) => _ref.Tell(new Request(n));

/// <summary>
/// TBD
/// </summary>
public void Cancel() => _ref.Tell(Actors.Cancel.Instance);
}

/// <summary>
/// TBD
/// </summary>
public sealed class OnErrorBlock
{
/// <summary>
/// TBD
/// </summary>
public readonly Exception Cause;
/// <summary>
/// TBD
/// </summary>
public readonly bool Stop;

/// <summary>
/// TBD
/// </summary>
/// <param name="cause">TBD</param>
/// <param name="stop">TBD</param>
public OnErrorBlock(Exception cause, bool stop)
{
Cause = cause;
Stop = stop;
}
}

/// <summary>
/// TBD
/// </summary>
internal class ActorPublisherState : ExtensionIdProvider<ActorPublisherState>, IExtension
{
/// <summary>
/// TBD
/// </summary>
public sealed class State
{
/// <summary>
/// TBD
/// </summary>
public readonly IUntypedSubscriber Subscriber;
/// <summary>
/// TBD
/// </summary>
public readonly long Demand;
/// <summary>
/// TBD
/// </summary>
public readonly LifecycleState LifecycleState;

/// <summary>
/// TBD
/// </summary>
/// <param name="subscriber">TBD</param>
/// <param name="demand">TBD</param>
/// <param name="lifecycleState">TBD</param>
public State(IUntypedSubscriber subscriber, long demand, LifecycleState lifecycleState)
{
Subscriber = subscriber;
Expand All @@ -600,24 +753,47 @@ public State(IUntypedSubscriber subscriber, long demand, LifecycleState lifecycl

private readonly ConcurrentDictionary<IActorRef, State> _state = new ConcurrentDictionary<IActorRef, State>();

/// <summary>
/// TBD
/// </summary>
public static readonly ActorPublisherState Instance = new ActorPublisherState();

private ActorPublisherState() { }

/// <summary>
/// TBD
/// </summary>
/// <param name="actorRef">TBD</param>
/// <returns>TBD</returns>
public State Get(IActorRef actorRef)
{
State state;
return _state.TryGetValue(actorRef, out state) ? state : null;
}

/// <summary>
/// TBD
/// </summary>
/// <param name="actorRef">TBD</param>
/// <param name="s">TBD</param>
public void Set(IActorRef actorRef, State s) => _state.AddOrUpdate(actorRef, s, (@ref, oldState) => s);

/// <summary>
/// TBD
/// </summary>
/// <param name="actorRef">TBD</param>
/// <returns>TBD</returns>
public State Remove(IActorRef actorRef)
{
State s;
return _state.TryRemove(actorRef, out s) ? s : null;
}

/// <summary>
/// TBD
/// </summary>
/// <param name="system">TBD</param>
/// <returns>TBD</returns>
public override ActorPublisherState CreateExtension(ExtendedActorSystem system) => new ActorPublisherState();
}
}
Loading