Fix a large series of connect's return scope attributes (#73)

source/concurrency/fork.d

@@ -11,6 +11,8 @@ version (Posix)
 struct ForkSender {
   alias Value = void;
   static assert(models!(typeof(this), isSender));
+  alias Fun = void delegate() shared;
+  alias AfterFork = void delegate(int);
   static struct Operation(Receiver) {
     private {
       Executor executor;
@@ -79,6 +81,12 @@ struct ForkSender {
         }
       }
     }
+    this(Executor executor, Fun fun, Receiver receiver, AfterFork afterFork) {
+      this.executor = executor;
+      this.fun = fun;
+      this.receiver = receiver;
+      this.afterFork = afterFork;
+    }
     void start() @trusted nothrow {
       import concurrency.thread : executeInNewThread, executeAndWait;
       import concurrency.utils : closure;
@@ -94,7 +102,7 @@ struct ForkSender {
     this.fun = fun;
     this.afterFork = afterFork;
   }
-  auto connect(Receiver)(Receiver receiver) @safe {
+  auto connect(Receiver)(return Receiver receiver) @safe {
     return new Operation!Receiver(executor, fun, receiver, afterFork);
   }
   static void reinitThreadLocks() {

source/concurrency/nursery.d

@@ -145,8 +145,12 @@ class Nursery : StopSource {
       op.start();
   }
 
-  auto connect(Receiver)(return Receiver receiver) @trusted scope return {
-    return (cast(shared)this).connect(receiver);
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
+    return asShared.connect(receiver);
+  }
+
+  private shared(Nursery) asShared() @trusted return scope {
+    return cast(shared)this;
   }
 
   auto connect(Receiver)(return Receiver receiver) shared scope return @trusted {

source/concurrency/operations/completewithcancellation.d

@@ -29,7 +29,7 @@ struct CompleteWithCancellationSender(Sender) if (models!(Sender, isSender)) {
   static assert(is(Sender.Value == void), "Sender must produce void to be able to complete with cancellation.");
   alias Value = void;
   Sender sender;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     /// ensure NRVO
     auto op = sender.connect(CompleteWithCancellationReceiver!(Receiver)(receiver));
     return op;

source/concurrency/operations/completewitherror.d

@@ -50,7 +50,7 @@ struct CompleteWithErrorSender(Sender) if (models!(Sender, isSender)) {
   alias Value = Sender.Value;
   Sender sender;
   Throwable t;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     /// ensure NRVO
     auto op = sender.connect(CompleteWithErrorReceiver!(Receiver)(receiver, t));
     return op;

source/concurrency/operations/forwardon.d

@@ -40,7 +40,7 @@ struct ForwardOnSender(Sender, Scheduler) if (models!(Sender, isSender)) {
   alias Value = Sender.Value;
   Sender sender;
   Scheduler scheduler;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     alias R = ForwardOnReceiver!(Receiver, Sender.Value, Scheduler);
     // ensure NRVO
     auto op = sender.connect(R(receiver, scheduler));

source/concurrency/operations/ignoreerror.d

@@ -15,7 +15,7 @@ struct IESender(Sender) if (models!(Sender, isSender)) {
   static assert(models!(typeof(this), isSender));
   alias Value = Sender.Value;
   Sender sender;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = sender.connect(IEReceiver!(Sender.Value,Receiver)(receiver));
     return op;

source/concurrency/operations/oncompletion.d

@@ -41,7 +41,7 @@ struct OnCompletionSender(Sender, SideEffect) if (models!(Sender, isSender)) {
   alias Value = Sender.Value;
   Sender sender;
   SideEffect effect;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = sender.connect(OnCompletionReceiver!(Sender.Value, SideEffect, Receiver)(receiver, effect));
     return op;

source/concurrency/operations/onerror.d

@@ -48,7 +48,7 @@ struct OnErrorSender(Sender, SideEffect) if (models!(Sender, isSender)) {
   alias Value = Sender.Value;
   Sender sender;
   SideEffect effect;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = sender.connect(OnErrorReceiver!(Sender.Value, SideEffect, Receiver)(receiver, effect));
     return op;

source/concurrency/operations/onresult.d

@@ -61,7 +61,7 @@ struct OnResultSender(Sender, SideEffect) if (models!(Sender, isSender)) {
   alias Value = Sender.Value;
   Sender sender;
   SideEffect effect;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = sender.connect(OnResultReceiver!(Sender.Value, SideEffect, Receiver)(receiver, effect));
     return op;

source/concurrency/operations/ontermination.d

@@ -42,7 +42,7 @@ struct OnTerminationSender(Sender, SideEffect) if (models!(Sender, isSender)) {
   alias Value = Sender.Value;
   Sender sender;
   SideEffect effect;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = sender.connect(OnTerminationReceiver!(Sender.Value, SideEffect, Receiver)(receiver, effect));
     return op;

source/concurrency/operations/race.d

@@ -102,7 +102,7 @@ struct RaceSender(Senders...)
   alias Value = Result!(Senders);
   Senders senders;
   bool noDropouts; // if true then we fail the moment one contender does, otherwise we keep running until one finishes
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = RaceOp!(Receiver, Senders)(receiver, senders, noDropouts);
     return op;

source/concurrency/operations/repeat.d

@@ -55,7 +55,7 @@ struct RepeatSender(Sender) if (models!(Sender, isSender)) {
   static assert(models!(typeof(this), isSender));
   alias Value = Sender.Value;
   Sender sender;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = RepeatOp!(Receiver, Sender)(sender, receiver);
     return op;

source/concurrency/operations/retry.d

@@ -84,7 +84,7 @@ struct RetrySender(Sender, Logic) if (models!(Sender, isSender) && models!(Logic
   alias Value = Sender.Value;
   Sender sender;
   Logic logic;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = RetryOp!(Receiver, Sender, Logic)(sender, RetryReceiver!(Receiver, Sender, Logic)(sender, receiver, logic));
     return op;

source/concurrency/operations/retrywhen.d

@@ -94,7 +94,7 @@ struct RetryWhenSender(Sender, Logic) if (isRetryWhenLogic!Logic) {
   alias Value = Sender.Value;
   Sender sender;
   Logic logic;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = RetryWhenOp!(Sender, Receiver, Logic)(sender, receiver, logic);
     return op;

source/concurrency/operations/stopon.d

@@ -42,7 +42,7 @@ struct StopOn(Sender) if (models!(Sender, isSender)) {
   alias Value = Sender.Value;
   Sender sender;
   StopToken stopToken;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     alias R = StopOnReceiver!(Receiver, Sender.Value);
     // ensure NRVO
     auto op = sender.connect(R(receiver, stopToken));

source/concurrency/operations/stopwhen.d

@@ -44,7 +44,7 @@ struct StopWhenSender(Sender, Trigger) if (models!(Sender, isSender) && models!(
   alias Value = Sender.Value;
   Sender sender;
   Trigger trigger;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = StopWhenOp!(Receiver, Sender, Trigger)(receiver, sender, trigger);
     return op;

source/concurrency/operations/then.d

@@ -58,7 +58,7 @@ struct ThenSender(Sender, Fun) if (models!(Sender, isSender)) {
   alias Value = ReturnType!fun;
   Sender sender;
   Fun fun;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     alias R = ThenReceiver!(Receiver, Sender.Value, Fun);
     // ensure NRVO
     auto op = sender.connect(R(receiver, fun));

source/concurrency/operations/toshared.d

@@ -96,7 +96,7 @@ class SharedSender(Sender, Scheduler, ResetLogic resetLogic) if (models!(Sender,
     this.sender = sender;
     this.scheduler = scheduler;
   }
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = SharedSenderOp!(Sender, Scheduler, resetLogic, Receiver)(this, receiver);
     return op;

source/concurrency/operations/via.d

@@ -75,7 +75,7 @@ struct ViaSender(SenderA, SenderB) if (models!(SenderA, isSender) && models!(Sen
     import std.typecons : Tuple;
     alias Value = Tuple!Values;
   }
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = senderB.connect(ViaBReceiver!(SenderA, SenderB.Value, Receiver)(senderA, receiver));
     return op;

source/concurrency/operations/whenall.d

@@ -153,7 +153,7 @@ struct WhenAllSender(Senders...)
   else
     alias Value = void;
   Senders senders;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = WhenAllOp!(Receiver, Senders)(receiver, senders);
     return op;

source/concurrency/operations/withchild.d

@@ -16,7 +16,7 @@ struct WithChildSender(SenderParent, SenderChild) if (models!(SenderParent, isSe
   alias Value = void;
   SenderParent a;
   SenderChild b;
-  auto connect(Receiver)(return Receiver receiver) @trusted scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     import concurrency.operations.whenall;
     import concurrency.operations.stopon;
     // ensure NRVO

source/concurrency/operations/withscheduler.d

@@ -41,7 +41,7 @@ struct WithSchedulerSender(Sender, Scheduler) if (models!(Sender, isSender)) {
   alias Value = Sender.Value;
   Sender sender;
   Scheduler scheduler;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     alias R = WithSchedulerReceiver!(Receiver, Sender.Value, Scheduler);
     // ensure NRVO
     auto op = sender.connect(R(receiver, scheduler));

source/concurrency/operations/withstopsource.d

@@ -43,7 +43,7 @@ private struct SSReceiver(Receiver, Value) {
     resetStopCallback();
     receiver.setError(e);
   }
-  auto getStopToken() nothrow @trusted {
+  auto getStopToken() nothrow @trusted scope {
     import core.atomic;
     if (this.combinedSource is null) {
       auto local = new StopSource();
@@ -77,7 +77,7 @@ struct SSSender(Sender) if (models!(Sender, isSender)) {
   alias Value = Sender.Value;
   Sender sender;
   StopSource stopSource;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     alias R = SSReceiver!(Receiver, Sender.Value);
     // ensure NRVO
     auto op = sender.connect(R(receiver, stopSource));

source/concurrency/operations/withstoptoken.d

@@ -20,6 +20,10 @@ auto withStopToken(Sender, Fun)(Sender sender, Fun fun) if (isThreadSafeFunction
 private struct STReceiver(Receiver, Value, Fun) {
   Receiver receiver;
   Fun fun;
+  this(return scope Receiver receiver, return Fun fun) @safe return scope {
+    this.receiver = receiver;
+    this.fun = fun;
+  }
   static if (is(Value == void)) {
     void setValue() @safe {
       static if (is(ReturnType!Fun == void)) {
@@ -67,7 +71,7 @@ struct STSender(Sender, Fun) if (models!(Sender, isSender)) {
   alias Value = ReturnType!fun;
   Sender sender;
   Fun fun;
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe return scope {
     alias R = STReceiver!(Receiver, Sender.Value, Fun);
     // ensure NRVO
     auto op = sender.connect(R(receiver, fun));

source/concurrency/scheduler.d

@@ -84,7 +84,7 @@ struct SchedulerAdapter(Worker) {
     static struct ScheduleSender {
       alias Value = void;
       Worker worker;
-      auto connect(Receiver)(return Receiver receiver) @safe scope return {
+      auto connect(Receiver)(return Receiver receiver) @safe return scope {
         // ensure NRVO
         auto op = ScheduleOp!(Receiver)(worker, receiver);
         return op;
@@ -296,7 +296,7 @@ struct ScheduleAfter {
   static assert (models!(typeof(this), isSender));
   alias Value = void;
   Duration duration;
-  auto connect(Receiver)(return Receiver receiver) @trusted scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe {
     // ensure NRVO
     auto op = receiver.getScheduler.scheduleAfter(duration).connect(receiver);
     return op;
@@ -306,7 +306,7 @@ struct ScheduleAfter {
 struct Schedule {
   static assert (models!(typeof(this), isSender));
   alias Value = void;
-  auto connect(Receiver)(return Receiver receiver) @trusted scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe {
     // ensure NRVO
     auto op = receiver.getScheduler.schedule().connect(receiver);
     return op;

source/concurrency/sender.d

@@ -159,7 +159,7 @@ struct JustFromSender(Fun) {
     }
   }
   Fun fun;
-  Op!Receiver connect(Receiver)(return Receiver receiver) @safe scope return {
+  Op!Receiver connect(Receiver)(return Receiver receiver) @safe return scope {
     // ensure NRVO
     auto op = Op!(Receiver)(receiver, fun);
     return op;
@@ -277,7 +277,7 @@ struct ThrowingSender {
       receiver.setError(new Exception("ThrowingSender"));
     }
   }
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe {
     // ensure NRVO
     auto op = Op!Receiver(receiver);
     return op;
@@ -296,7 +296,7 @@ struct DoneSender {
       receiver.setDone();
     }
   }
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe {
     // ensure NRVO
     auto op = DoneOp!(Receiver)(receiver);
     return op;
@@ -316,7 +316,7 @@ struct VoidSender {
       receiver.setValueOrError();
     }
   }
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe {
     // ensure NRVO
     auto op = VoidOp!Receiver(receiver);
     return op;
@@ -337,7 +337,7 @@ struct ErrorSender {
       receiver.setError(exception);
     }
   }
-  auto connect(Receiver)(return Receiver receiver) @safe scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe {
     // ensure NRVO
     auto op = ErrorOp!(Receiver)(receiver, exception);
     return op;
@@ -367,7 +367,7 @@ template OpType(Sender, Receiver) {
 struct DelaySender {
   alias Value = void;
   Duration dur;
-  auto connect(Receiver)(return Receiver receiver) @trusted scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe {
     // ensure NRVO
     auto op = receiver.getScheduler().scheduleAfter(dur).connect(receiver);
     return op;
@@ -590,7 +590,7 @@ struct PromiseSender(T) {
     auto op = (cast(shared)this).connect(receiver);
     return op;
   }
-  auto connect(Receiver)(return Receiver receiver) @safe shared scope return {
+  auto connect(Receiver)(return Receiver receiver) @safe shared return scope {
     // ensure NRVO
     auto op = PromiseSenderOp!(T, Receiver)(promise, receiver);
     return op;

source/concurrency/stream/flatmapbase.d

@@ -28,7 +28,7 @@ template FlatMapBaseStreamOp(Stream, Fun, OnOverlap overlap) {
     State state;
     @disable this(ref return scope typeof(this) rhs);
     @disable this(this);
-    this(Stream stream, Fun fun, return DG dg, return Receiver receiver) @trusted scope return {
+    this(Stream stream, Fun fun, return DG dg, return Receiver receiver) @trusted return scope {
       this.fun = fun;
       state = new State(dg, receiver);
       // TODO: would it be good to do the fun in a transform operation?

source/concurrency/stream/package.d

@@ -182,7 +182,7 @@ auto intervalStream(Duration duration, bool emitAtStart = false) {
     Duration duration;
     DG dg;
     bool emitAtStart;
-    auto connect(Receiver)(return Receiver receiver) @safe scope return {
+    auto connect(Receiver)(return Receiver receiver) @safe return scope {
       // ensure NRVO
       auto op = Op!(Receiver)(duration, dg, receiver, emitAtStart);
       return op;
@@ -308,7 +308,7 @@ shared struct SharedStream(T) {
       alias Value = void;
       shared SharedStream!T source;
       DG dg;
-      auto connect(Receiver)(return Receiver receiver) @safe scope return {
+      auto connect(Receiver)(return Receiver receiver) @safe return scope {
         // ensure NRVO
         auto op = Op!(Receiver)(source, dg, receiver);
         return op;

source/concurrency/stream/stream.d

@@ -74,7 +74,7 @@ auto fromStreamOp(StreamElementType, SenderValue, alias Op, Args...)(Args args)
     alias Value = SenderValue;
     Args args;
     DG dg;
-    auto connect(Receiver)(return Receiver receiver) @safe scope return {
+    auto connect(Receiver)(return Receiver receiver) @safe return scope {
       // ensure NRVO
       auto op = Op!(Receiver)(args, dg, receiver);
       return op;
@@ -130,7 +130,7 @@ template loopStream(E) {
         alias Value = void;
         T t;
         DG dg;
-        auto connect(Receiver)(return Receiver receiver) @safe scope return {
+        auto connect(Receiver)(return Receiver receiver) @safe return scope {
           // ensure NRVO
           auto op = LoopOp!(Receiver)(t, dg, receiver);
           return op;