diff --git a/sandbox/MauiApp1/MainPage.xaml b/sandbox/MauiApp1/MainPage.xaml
index 13462f1b..06eba702 100644
--- a/sandbox/MauiApp1/MainPage.xaml
+++ b/sandbox/MauiApp1/MainPage.xaml
@@ -1,6 +1,8 @@
 <?xml version="1.0" encoding="utf-8" ?>
 <ContentPage xmlns="http://schemas.microsoft.com/dotnet/2021/maui"
              xmlns:x="http://schemas.microsoft.com/winfx/2009/xaml"
+             xmlns:mauiApp1="clr-namespace:MauiApp1"
+             x:DataType="mauiApp1:BasicUsagesViewModel"
              x:Class="MauiApp1.MainPage">
 
     <ScrollView>
@@ -36,8 +38,11 @@
                 x:Name="CounterBtn"
                 Text="Click me"
                 SemanticProperties.Hint="Counts the number of times you click"
-                Clicked="OnCounterClicked"
+                Command="{Binding DoProcessingCommand}"
                 HorizontalOptions="Fill" />
+
+            <Label Text="Commands Are Executing?" />
+            <Entry Text="{Binding IsProcessing.Value, Mode=OneWay, StringFormat='{0}'}" />
         </VerticalStackLayout>
     </ScrollView>
 
diff --git a/sandbox/MauiApp1/MainPage.xaml.cs b/sandbox/MauiApp1/MainPage.xaml.cs
index 4e2dac55..e604bec8 100644
--- a/sandbox/MauiApp1/MainPage.xaml.cs
+++ b/sandbox/MauiApp1/MainPage.xaml.cs
@@ -1,4 +1,5 @@
-using R3;
+using System.Diagnostics;
+using R3;
 
 namespace MauiApp1;
 
@@ -7,15 +8,35 @@ public class BasicUsagesViewModel : IDisposable
     public BindableReactiveProperty<string> Input { get; }
     public BindableReactiveProperty<string> Output { get; }
 
+    public ReactiveCommand<Unit> DoProcessingCommand { get; }
+
+    public IReadOnlyBindableReactiveProperty<bool> IsProcessing { get; }
+
     public BasicUsagesViewModel()
     {
         Input = new BindableReactiveProperty<string>("");
         Output = Input.Select(x => x.ToUpper()).ToBindableReactiveProperty("");
+
+        DoProcessingCommand = new(DoProcessingAsync, AwaitOperation.Parallel);
+
+        IsProcessing =
+            this.DoProcessingCommand
+                .IsExecuting
+                .ToReadOnlyBindableReactiveProperty(false);
+    }
+
+    private async ValueTask DoProcessingAsync(Unit input, CancellationToken token)
+    {
+        var now = DateTime.Now;
+
+        Debug.WriteLine($"Starting processing of input: {now}");
+        await Task.Delay(TimeSpan.FromSeconds(10), token);
+        Debug.WriteLine($"Finished processing of input: {now}");
     }
 
     public void Dispose()
     {
-        Disposable.Dispose(Input, Output);
+        Disposable.Dispose(Input, Output, DoProcessingCommand, IsProcessing);
     }
 }
 
@@ -27,14 +48,6 @@ public MainPage()
     {
         InitializeComponent();
 
-        var startDate = DateTime.Now;
-        Observable.Interval(TimeSpan.FromMilliseconds(500))
-            .Subscribe(x =>
-            {
-                throw new InvalidOperationException("oppeke");
-                Label1.Text = $"Timer: {(DateTime.Now - startDate).TotalMilliseconds}";
-            });
-
         var frameCount = 0;
         Observable.IntervalFrame(1)
             .Subscribe(x =>
@@ -45,8 +58,11 @@ public MainPage()
         this.Loaded += (sender, args) =>
         {
             var viewModel = new BasicUsagesViewModel();
+
             BindingContext = viewModel;
         };
+
+
     }
 
     void OnCounterClicked(object sender, EventArgs e)
diff --git a/sandbox/MauiApp1/MauiApp1.csproj b/sandbox/MauiApp1/MauiApp1.csproj
index 9aa99769..e9916d3b 100644
--- a/sandbox/MauiApp1/MauiApp1.csproj
+++ b/sandbox/MauiApp1/MauiApp1.csproj
@@ -32,8 +32,8 @@
         <ApplicationDisplayVersion>1.0</ApplicationDisplayVersion>
         <ApplicationVersion>1</ApplicationVersion>
 
-        <SupportedOSPlatformVersion Condition="$([MSBuild]::GetTargetPlatformIdentifier('$(TargetFramework)')) == 'ios'">11.0</SupportedOSPlatformVersion>
-        <SupportedOSPlatformVersion Condition="$([MSBuild]::GetTargetPlatformIdentifier('$(TargetFramework)')) == 'maccatalyst'">13.1</SupportedOSPlatformVersion>
+        <SupportedOSPlatformVersion Condition="$([MSBuild]::GetTargetPlatformIdentifier('$(TargetFramework)')) == 'ios'">14.0</SupportedOSPlatformVersion>
+        <SupportedOSPlatformVersion Condition="$([MSBuild]::GetTargetPlatformIdentifier('$(TargetFramework)')) == 'maccatalyst'">15.0</SupportedOSPlatformVersion>
         <SupportedOSPlatformVersion Condition="$([MSBuild]::GetTargetPlatformIdentifier('$(TargetFramework)')) == 'android'">21.0</SupportedOSPlatformVersion>
         <SupportedOSPlatformVersion Condition="$([MSBuild]::GetTargetPlatformIdentifier('$(TargetFramework)')) == 'windows'">10.0.17763.0</SupportedOSPlatformVersion>
         <TargetPlatformMinVersion Condition="$([MSBuild]::GetTargetPlatformIdentifier('$(TargetFramework)')) == 'windows'">10.0.17763.0</TargetPlatformMinVersion>
diff --git a/src/R3/Operators/SelectAwait.cs b/src/R3/Operators/SelectAwait.cs
index c20ab8cb..48a761ae 100644
--- a/src/R3/Operators/SelectAwait.cs
+++ b/src/R3/Operators/SelectAwait.cs
@@ -9,6 +9,12 @@ public static Observable<TResult> SelectAwait<T, TResult>(this Observable<T> sou
     {
         return new SelectAwait<T, TResult>(source, selector, awaitOperation, configureAwait, cancelOnCompleted, maxConcurrent);
     }
+
+    /// <param name="maxConcurrent">This option is only valid for AwaitOperation.Parallel and AwaitOperation.SequentialParallel. It sets the number of concurrent executions. If set to -1, there is no limit.</param>
+    public static Observable<TResult> SelectAwait<T, TResult, TState>(this Observable<T> source, TState state, Func<T, TState, CancellationToken, ValueTask<TResult>> selector, AwaitOperation awaitOperation = AwaitOperation.Sequential, bool configureAwait = true, bool cancelOnCompleted = false, int maxConcurrent = -1)
+    {
+        return new SelectAwait<T, TResult, TState>(source, state, selector, awaitOperation, configureAwait, cancelOnCompleted, maxConcurrent);
+    }
 }
 
 internal sealed class SelectAwait<T, TResult>(Observable<T> source, Func<T, CancellationToken, ValueTask<TResult>> selector, AwaitOperation awaitOperation, bool configureAwait, bool cancelOnCompleted, int maxConcurrent) : Observable<TResult>
@@ -274,3 +280,267 @@ protected override void PublishOnCompleted(Result result)
         }
     }
 }
+
+internal sealed class SelectAwait<T, TResult, TState>(Observable<T> source, TState state, Func<T, TState, CancellationToken, ValueTask<TResult>> selector, AwaitOperation awaitOperation, bool configureAwait, bool cancelOnCompleted, int maxConcurrent) : Observable<TResult>
+{
+    protected override IDisposable SubscribeCore(Observer<TResult> observer)
+    {
+        switch (awaitOperation)
+        {
+            case AwaitOperation.Sequential:
+                return source.Subscribe(new SelectAwaitSequential(observer, state, selector, configureAwait, cancelOnCompleted));
+            case AwaitOperation.Drop:
+                return source.Subscribe(new SelectAwaitDrop(observer, state, selector, configureAwait, cancelOnCompleted));
+            case AwaitOperation.Switch:
+                return source.Subscribe(new SelectAwaitSwitch(observer, state, selector, configureAwait, cancelOnCompleted));
+            case AwaitOperation.Parallel:
+                if (maxConcurrent == -1)
+                {
+                    return source.Subscribe(new SelectAwaitParallel(observer, state, selector, configureAwait, cancelOnCompleted));
+                }
+                else
+                {
+                    if (maxConcurrent == 0 || maxConcurrent < -1) throw new ArgumentException("maxConcurrent must be a -1 or greater than 1.");
+                    return source.Subscribe(new SelectAwaitParallelConcurrentLimit(observer, state, selector, configureAwait, cancelOnCompleted, maxConcurrent));
+                }
+            case AwaitOperation.SequentialParallel:
+                if (maxConcurrent == -1)
+                {
+                    return source.Subscribe(new SelectAwaitSequentialParallel(observer, state, selector, configureAwait, cancelOnCompleted));
+                }
+                else
+                {
+                    if (maxConcurrent == 0 || maxConcurrent < -1) throw new ArgumentException("maxConcurrent must be a -1 or greater than 1.");
+                    return source.Subscribe(new SelectAwaitSequentialParallelConcurrentLimit(observer, state, selector, configureAwait, cancelOnCompleted, maxConcurrent));
+                }
+            case AwaitOperation.ThrottleFirstLast:
+                return source.Subscribe(new SelectAwaitThrottleFirstLast(observer, state, selector, configureAwait, cancelOnCompleted));
+            default:
+                throw new ArgumentException();
+        }
+    }
+
+    sealed class SelectAwaitSequential(Observer<TResult> observer, TState state, Func<T, TState, CancellationToken, ValueTask<TResult>> selector, bool configureAwait, bool cancelOnCompleted)
+        : AwaitOperationSequentialObserver<T>(configureAwait, cancelOnCompleted)
+    {
+#if NET6_0_OR_GREATER
+        [AsyncMethodBuilderAttribute(typeof(PoolingAsyncValueTaskMethodBuilder))]
+#endif
+        protected override async ValueTask OnNextAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+        {
+            var v = await selector(value, state, cancellationToken).ConfigureAwait(configureAwait);
+            observer.OnNext(v);
+        }
+
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            observer.OnErrorResume(error);
+        }
+
+        protected override void PublishOnCompleted(Result result)
+        {
+            observer.OnCompleted(result);
+        }
+    }
+
+    sealed class SelectAwaitDrop(Observer<TResult> observer, TState state, Func<T, TState, CancellationToken, ValueTask<TResult>> selector, bool configureAwait, bool cancelOnCompleted)
+        : AwaitOperationDropObserver<T>(configureAwait, cancelOnCompleted)
+    {
+#if NET6_0_OR_GREATER
+        [AsyncMethodBuilderAttribute(typeof(PoolingAsyncValueTaskMethodBuilder))]
+#endif
+        protected override async ValueTask OnNextAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+        {
+            var v = await selector(value, state, cancellationToken).ConfigureAwait(configureAwait);
+            observer.OnNext(v);
+        }
+
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            observer.OnErrorResume(error);
+        }
+
+        protected override void PublishOnCompleted(Result result)
+        {
+            observer.OnCompleted(result);
+        }
+    }
+
+    sealed class SelectAwaitParallel(Observer<TResult> observer, TState state, Func<T, TState, CancellationToken, ValueTask<TResult>> selector, bool configureAwait, bool cancelOnCompleted)
+        : AwaitOperationParallelObserver<T>(configureAwait, cancelOnCompleted)
+    {
+
+#if NET6_0_OR_GREATER
+        [AsyncMethodBuilderAttribute(typeof(PoolingAsyncValueTaskMethodBuilder))]
+#endif
+        protected override async ValueTask OnNextAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+        {
+            var v = await selector(value, state, cancellationToken).ConfigureAwait(configureAwait);
+            lock (gate)
+            {
+                observer.OnNext(v);
+            }
+        }
+
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            lock (gate)
+            {
+                observer.OnErrorResume(error);
+            }
+        }
+
+        protected override void PublishOnCompleted(Result result)
+        {
+            lock (gate)
+            {
+                observer.OnCompleted(result);
+            }
+        }
+    }
+
+
+    sealed class SelectAwaitSwitch(Observer<TResult> observer, TState state, Func<T, TState, CancellationToken, ValueTask<TResult>> selector, bool configureAwait, bool cancelOnCompleted)
+        : AwaitOperationSwitchObserver<T>(configureAwait, cancelOnCompleted)
+    {
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            lock (gate)
+            {
+                observer.OnErrorResume(error);
+            }
+        }
+
+        protected override async ValueTask OnNextAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+        {
+            var v = await selector(value, state, cancellationToken).ConfigureAwait(configureAwait);
+            lock (gate)
+            {
+                if (!cancellationToken.IsCancellationRequested)
+                {
+                    observer.OnNext(v);
+                }
+            }
+        }
+
+        protected override void PublishOnCompleted(Result result)
+        {
+            lock (gate)
+            {
+                observer.OnCompleted(result);
+            }
+        }
+    }
+
+    sealed class SelectAwaitSequentialParallel(Observer<TResult> observer, TState state, Func<T, TState, CancellationToken, ValueTask<TResult>> selector, bool configureAwait, bool cancelOnCompleted)
+        : AwaitOperationSequentialParallelObserver<T, TResult>(configureAwait, cancelOnCompleted)
+    {
+
+#if NET6_0_OR_GREATER
+        [AsyncMethodBuilderAttribute(typeof(PoolingAsyncValueTaskMethodBuilder))]
+#endif
+        protected override ValueTask<TResult> OnNextTaskAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+        {
+            return selector(value, state, cancellationToken);
+        }
+
+        protected override void PublishOnNext(T _, TResult result)
+        {
+            observer.OnNext(result);
+        }
+
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            observer.OnErrorResume(error);
+        }
+
+        protected override void PublishOnCompleted(Result result)
+        {
+            observer.OnCompleted(result);
+        }
+    }
+
+    sealed class SelectAwaitParallelConcurrentLimit(Observer<TResult> observer, TState state, Func<T, TState, CancellationToken, ValueTask<TResult>> selector, bool configureAwait, bool cancelOnCompleted, int maxConcurrent)
+        : AwaitOperationParallelConcurrentLimitObserver<T>(configureAwait, cancelOnCompleted, maxConcurrent)
+    {
+
+#if NET6_0_OR_GREATER
+        [AsyncMethodBuilderAttribute(typeof(PoolingAsyncValueTaskMethodBuilder))]
+#endif
+        protected override async ValueTask OnNextAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+        {
+            var v = await selector(value, state, cancellationToken).ConfigureAwait(configureAwait);
+            lock (gate)
+            {
+                observer.OnNext(v);
+            }
+        }
+
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            lock (gate)
+            {
+                observer.OnErrorResume(error);
+            }
+        }
+
+        protected override void PublishOnCompleted(Result result)
+        {
+            lock (gate)
+            {
+                observer.OnCompleted(result);
+            }
+        }
+    }
+
+    sealed class SelectAwaitSequentialParallelConcurrentLimit(Observer<TResult> observer, TState state, Func<T, TState, CancellationToken, ValueTask<TResult>> selector, bool configureAwait, bool cancelOnCompleted, int maxConcurrent)
+        : AwaitOperationSequentialParallelConcurrentLimitObserver<T, TResult>(configureAwait, cancelOnCompleted, maxConcurrent)
+    {
+
+#if NET6_0_OR_GREATER
+        [AsyncMethodBuilderAttribute(typeof(PoolingAsyncValueTaskMethodBuilder))]
+#endif
+        protected override ValueTask<TResult> OnNextTaskAsyncCore(T value, CancellationToken cancellationToken, bool configureAwait)
+        {
+            return selector(value, state, cancellationToken);
+        }
+
+        protected override void PublishOnNext(T _, TResult result)
+        {
+            observer.OnNext(result);
+        }
+
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            observer.OnErrorResume(error);
+        }
+
+        protected override void PublishOnCompleted(Result result)
+        {
+            observer.OnCompleted(result);
+        }
+    }
+
+    sealed class SelectAwaitThrottleFirstLast(Observer<TResult> observer, TState state, Func<T, TState, CancellationToken, ValueTask<TResult>> selector, bool configureAwait, bool cancelOnCompleted)
+        : AwaitOperationThrottleFirstLastObserver<T>(configureAwait, cancelOnCompleted)
+    {
+#if NET6_0_OR_GREATER
+        [AsyncMethodBuilderAttribute(typeof(PoolingAsyncValueTaskMethodBuilder))]
+#endif
+        protected override async ValueTask OnNextAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+        {
+            var v = await selector(value, state, cancellationToken).ConfigureAwait(configureAwait);
+            observer.OnNext(v);
+        }
+
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            observer.OnErrorResume(error);
+        }
+
+        protected override void PublishOnCompleted(Result result)
+        {
+            observer.OnCompleted(result);
+        }
+    }
+}
diff --git a/src/R3/Operators/SubscribeAwait.cs b/src/R3/Operators/SubscribeAwait.cs
index 8563573e..d8a63460 100644
--- a/src/R3/Operators/SubscribeAwait.cs
+++ b/src/R3/Operators/SubscribeAwait.cs
@@ -46,6 +46,47 @@ public static IDisposable SubscribeAwait<T>(this Observable<T> source, Func<T, C
                 throw new ArgumentException();
         }
     }
+
+    /// <param name="maxConcurrent">This option is only valid for AwaitOperation.Parallel and AwaitOperation.SequentialParallel. It sets the number of concurrent executions. If set to -1, there is no limit.</param>
+    public static IDisposable SubscribeAwait<T, TState>(this Observable<T> source, TState state, Func<T, TState, CancellationToken, ValueTask> onNextAsync, AwaitOperation awaitOperation = AwaitOperation.Sequential, bool configureAwait = true, bool cancelOnCompleted = false, int maxConcurrent = -1)
+    {
+        return SubscribeAwait(source, state, onNextAsync, Stubs<TState>.HandleException, Stubs<TState>.HandleResult, awaitOperation, configureAwait, cancelOnCompleted, maxConcurrent);
+    }
+
+    /// <param name="maxConcurrent">This option is only valid for AwaitOperation.Parallel and AwaitOperation.SequentialParallel. It sets the number of concurrent executions. If set to -1, there is no limit.</param>
+    public static IDisposable SubscribeAwait<T, TState>(this Observable<T> source, TState state, Func<T, TState, CancellationToken, ValueTask> onNextAsync, Action<Result, TState> onCompleted, AwaitOperation awaitOperation = AwaitOperation.Sequential, bool configureAwait = true, bool cancelOnCompleted = false, int maxConcurrent = -1)
+    {
+        return SubscribeAwait(source, state, onNextAsync, Stubs<TState>.HandleException, onCompleted, awaitOperation, configureAwait, cancelOnCompleted, maxConcurrent);
+    }
+
+    public static IDisposable SubscribeAwait<T, TState>(this Observable<T> source, TState state, Func<T, TState, CancellationToken, ValueTask> onNextAsync, Action<Exception, TState> onErrorResume, Action<Result, TState> onCompleted, AwaitOperation awaitOperation = AwaitOperation.Sequential, bool configureAwait = true, bool cancelOnCompleted = false, int maxConcurrent = -1)
+    {
+        switch (awaitOperation)
+        {
+            case AwaitOperation.Sequential:
+                return source.Subscribe(new SubscribeAwaitSequential<T, TState>(state, onNextAsync, onErrorResume, onCompleted, configureAwait, cancelOnCompleted));
+            case AwaitOperation.Drop:
+                return source.Subscribe(new SubscribeAwaitDrop<T, TState>(state, onNextAsync, onErrorResume, onCompleted, configureAwait, cancelOnCompleted));
+            case AwaitOperation.Parallel:
+                if (maxConcurrent == -1)
+                {
+                    return source.Subscribe(new SubscribeAwaitParallel<T, TState>(state, onNextAsync, onErrorResume, onCompleted, configureAwait, cancelOnCompleted));
+                }
+                else
+                {
+                    if (maxConcurrent == 0 || maxConcurrent < -1) throw new ArgumentException("maxConcurrent must be a -1 or greater than 1.");
+                    return source.Subscribe(new SubscribeAwaitParallelConcurrentLimit<T, TState>(state, onNextAsync, onErrorResume, onCompleted, configureAwait, cancelOnCompleted, maxConcurrent));
+                }
+            case AwaitOperation.Switch:
+                return source.Subscribe(new SubscribeAwaitSwitch<T, TState>(state, onNextAsync, onErrorResume, onCompleted, configureAwait, cancelOnCompleted));
+            case AwaitOperation.SequentialParallel:
+                throw new ArgumentException("SubscribeAwait does not support SequentialParallel. Use Sequential for sequential operation, use parallel for parallel operation instead.");
+            case AwaitOperation.ThrottleFirstLast:
+                return source.Subscribe(new SubscribeAwaitThrottleFirstLast<T, TState>(state, onNextAsync, onErrorResume, onCompleted, configureAwait, cancelOnCompleted));
+            default:
+                throw new ArgumentException();
+        }
+    }
 }
 
 internal sealed class SubscribeAwaitSequential<T>(Func<T, CancellationToken, ValueTask> onNextAsync, Action<Exception> onErrorResume, Action<Result> onCompleted, bool configureAwait, bool cancelOnCompleted)
@@ -67,6 +108,25 @@ protected override void PublishOnCompleted(Result result)
     }
 }
 
+internal sealed class SubscribeAwaitSequential<T, TState>(TState state, Func<T, TState, CancellationToken, ValueTask> onNextAsync, Action<Exception, TState> onErrorResume, Action<Result, TState> onCompleted, bool configureAwait, bool cancelOnCompleted)
+    : AwaitOperationSequentialObserver<T>(configureAwait, cancelOnCompleted)
+{
+    protected override ValueTask OnNextAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+    {
+        return onNextAsync(value, state, cancellationToken);
+    }
+
+    protected override void OnErrorResumeCore(Exception error)
+    {
+        onErrorResume(error, state);
+    }
+
+    protected override void PublishOnCompleted(Result result)
+    {
+        onCompleted(result, state);
+    }
+}
+
 internal sealed class SubscribeAwaitDrop<T>(Func<T, CancellationToken, ValueTask> onNextAsync, Action<Exception> onErrorResume, Action<Result> onCompleted, bool configureAwait, bool cancelOnCompleted)
     : AwaitOperationDropObserver<T>(configureAwait, cancelOnCompleted)
 {
@@ -86,6 +146,25 @@ protected override void PublishOnCompleted(Result result)
     }
 }
 
+internal sealed class SubscribeAwaitDrop<T, TState>(TState state, Func<T, TState, CancellationToken, ValueTask> onNextAsync, Action<Exception, TState> onErrorResume, Action<Result, TState> onCompleted, bool configureAwait, bool cancelOnCompleted)
+    : AwaitOperationDropObserver<T>(configureAwait, cancelOnCompleted)
+{
+    protected override ValueTask OnNextAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+    {
+        return onNextAsync(value, state, cancellationToken);
+    }
+
+    protected override void OnErrorResumeCore(Exception error)
+    {
+        onErrorResume(error, state);
+    }
+
+    protected override void PublishOnCompleted(Result result)
+    {
+        onCompleted(result, state);
+    }
+}
+
 sealed class SubscribeAwaitParallel<T>(Func<T, CancellationToken, ValueTask> onNextAsync, Action<Exception> onErrorResume, Action<Result> onCompleted, bool configureAwait, bool cancelOnCompleted)
     : AwaitOperationParallelObserver<T>(configureAwait, cancelOnCompleted)
 {
@@ -111,6 +190,31 @@ protected override void PublishOnCompleted(Result result)
     }
 }
 
+sealed class SubscribeAwaitParallel<T, TState>(TState state, Func<T, TState, CancellationToken, ValueTask> onNextAsync, Action<Exception, TState> onErrorResume, Action<Result, TState> onCompleted, bool configureAwait, bool cancelOnCompleted)
+    : AwaitOperationParallelObserver<T>(configureAwait, cancelOnCompleted)
+{
+    protected override ValueTask OnNextAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+    {
+        return onNextAsync(value, state, cancellationToken);
+    }
+
+    protected override void OnErrorResumeCore(Exception error)
+    {
+        lock (gate)
+        {
+            onErrorResume(error, state);
+        }
+    }
+
+    protected override void PublishOnCompleted(Result result)
+    {
+        lock (gate)
+        {
+            onCompleted(result, state);
+        }
+    }
+}
+
 sealed class SubscribeAwaitSwitch<T>(Func<T, CancellationToken, ValueTask> onNextAsync, Action<Exception> onErrorResume, Action<Result> onCompleted, bool configureAwait, bool cancelOnCompleted)
     : AwaitOperationSwitchObserver<T>(configureAwait, cancelOnCompleted)
 {
@@ -136,6 +240,31 @@ protected override void PublishOnCompleted(Result result)
     }
 }
 
+sealed class SubscribeAwaitSwitch<T, TState>(TState state, Func<T, TState, CancellationToken, ValueTask> onNextAsync, Action<Exception, TState> onErrorResume, Action<Result, TState> onCompleted, bool configureAwait, bool cancelOnCompleted)
+    : AwaitOperationSwitchObserver<T>(configureAwait, cancelOnCompleted)
+{
+    protected override ValueTask OnNextAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+    {
+        return onNextAsync(value, state, cancellationToken);
+    }
+
+    protected override void OnErrorResumeCore(Exception error)
+    {
+        lock (gate)
+        {
+            onErrorResume(error, state);
+        }
+    }
+
+    protected override void PublishOnCompleted(Result result)
+    {
+        lock (gate)
+        {
+            onCompleted(result, state);
+        }
+    }
+}
+
 sealed class SubscribeAwaitParallelConcurrentLimit<T>(Func<T, CancellationToken, ValueTask> onNextAsync, Action<Exception> onErrorResume, Action<Result> onCompleted, bool configureAwait, bool cancelOnCompleted, int maxConcurrent)
     : AwaitOperationParallelConcurrentLimitObserver<T>(configureAwait, cancelOnCompleted, maxConcurrent)
 {
@@ -161,6 +290,31 @@ protected override void PublishOnCompleted(Result result)
     }
 }
 
+sealed class SubscribeAwaitParallelConcurrentLimit<T, TState>(TState state, Func<T, TState, CancellationToken, ValueTask> onNextAsync, Action<Exception, TState> onErrorResume, Action<Result, TState> onCompleted, bool configureAwait, bool cancelOnCompleted, int maxConcurrent)
+    : AwaitOperationParallelConcurrentLimitObserver<T>(configureAwait, cancelOnCompleted, maxConcurrent)
+{
+    protected override ValueTask OnNextAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+    {
+        return onNextAsync(value, state, cancellationToken);
+    }
+
+    protected override void OnErrorResumeCore(Exception error)
+    {
+        lock (gate)
+        {
+            onErrorResume(error, state);
+        }
+    }
+
+    protected override void PublishOnCompleted(Result result)
+    {
+        lock (gate)
+        {
+            onCompleted(result, state);
+        }
+    }
+}
+
 internal sealed class SubscribeAwaitThrottleFirstLast<T>(Func<T, CancellationToken, ValueTask> onNextAsync, Action<Exception> onErrorResume, Action<Result> onCompleted, bool configureAwait, bool cancelOnCompleted)
     : AwaitOperationThrottleFirstLastObserver<T>(configureAwait, cancelOnCompleted)
 {
@@ -179,3 +333,22 @@ protected override void PublishOnCompleted(Result result)
         onCompleted(result);
     }
 }
+
+internal sealed class SubscribeAwaitThrottleFirstLast<T, TState>(TState state, Func<T, TState, CancellationToken, ValueTask> onNextAsync, Action<Exception, TState> onErrorResume, Action<Result, TState> onCompleted, bool configureAwait, bool cancelOnCompleted)
+    : AwaitOperationThrottleFirstLastObserver<T>(configureAwait, cancelOnCompleted)
+{
+    protected override ValueTask OnNextAsync(T value, CancellationToken cancellationToken, bool configureAwait)
+    {
+        return onNextAsync(value, state, cancellationToken);
+    }
+
+    protected override void OnErrorResumeCore(Exception error)
+    {
+        onErrorResume(error, state);
+    }
+
+    protected override void PublishOnCompleted(Result result)
+    {
+        onCompleted(result, state);
+    }
+}
diff --git a/src/R3/ReactiveCommand.cs b/src/R3/ReactiveCommand.cs
index 89aeb8a4..20ca7f59 100644
--- a/src/R3/ReactiveCommand.cs
+++ b/src/R3/ReactiveCommand.cs
@@ -9,9 +9,13 @@ public class ReactiveCommand<T> : Observable<T>, ICommand, IDisposable
     CompleteState completeState;     // struct(int, IntPtr)
     IDisposable subscription; // from canExecuteSource (and onNext).
     bool canExecute; // set from observable sequence
+    int executionCount;
+    readonly object gate = new();
 
     public event EventHandler? CanExecuteChanged;
 
+    public ReactiveProperty<bool> IsExecuting { get; } = new();
+
     public ReactiveCommand()
     {
         this.list = new FreeListCore<Subscription>(this);
@@ -23,14 +27,14 @@ public ReactiveCommand(Action<T> execute)
     {
         this.list = new FreeListCore<Subscription>(this);
         this.canExecute = true;
-        this.subscription = this.Subscribe(execute);
+        this.subscription = this.Subscribe((HandleExecutionAction: (Action<T, Action<T>>)HandleExecution, Action: execute), static (value, state) => state.HandleExecutionAction(value, state.Action));
     }
 
     public ReactiveCommand(Func<T, CancellationToken, ValueTask> executeAsync, AwaitOperation awaitOperation = AwaitOperation.Sequential, bool configureAwait = true, bool cancelOnCompleted = false, int maxSequential = -1)
     {
         this.list = new FreeListCore<Subscription>(this);
         this.canExecute = true;
-        this.subscription = this.SubscribeAwait(executeAsync, awaitOperation, configureAwait, cancelOnCompleted, maxSequential);
+        this.subscription = this.SubscribeAwait((HandleAsyncExecutionFunc: (Func<T, Func<T, CancellationToken, ValueTask>, CancellationToken, ValueTask>)HandleAsyncExecution, Func: executeAsync), static (value, state, cancellationToken) => state.HandleAsyncExecutionFunc(value, state.Func, cancellationToken), awaitOperation, configureAwait, cancelOnCompleted, maxSequential);
     }
 
     public ReactiveCommand(Observable<bool> canExecuteSource, bool initialCanExecute)
@@ -89,6 +93,56 @@ internal void CombineSubscription(IDisposable disposable)
         this.subscription = Disposable.Combine(this.subscription, disposable);
     }
 
+    private void HandleExecution(T value, Action<T> execute)
+    {
+        try
+        {
+            lock (gate)
+            {
+                executionCount++;
+                IsExecuting.Value = executionCount > 0;
+            }
+
+            execute(value);
+        }
+        finally
+        {
+            lock (gate)
+            {
+                executionCount--;
+                if (executionCount == 0)
+                {
+                    IsExecuting.Value = false;
+                }
+            }
+        }
+    }
+
+    private async ValueTask HandleAsyncExecution(T value, Func<T, CancellationToken, ValueTask> executeAsync, CancellationToken cancellationToken)
+    {
+        try
+        {
+            lock (gate)
+            {
+                executionCount++;
+                IsExecuting.Value = executionCount > 0;
+            }
+
+            await executeAsync(value, cancellationToken);
+        }
+        finally
+        {
+            lock (gate)
+            {
+                executionCount--;
+                if (executionCount == 0)
+                {
+                    IsExecuting.Value = false;
+                }
+            }
+        }
+    }
+
     protected override IDisposable SubscribeCore(Observer<T> observer)
     {
         var result = completeState.TryGetResult();
@@ -130,6 +184,7 @@ public void Dispose(bool callOnCompleted)
                 }
             }
 
+            IsExecuting.Dispose();
             list.Dispose();
             subscription.Dispose();
         }
@@ -166,12 +221,16 @@ public class ReactiveCommand<TInput, TOutput> : Observable<TOutput>, ICommand, I
     CompleteState completeState;     // struct(int, IntPtr)
     bool canExecute; // set from observable sequence
     IDisposable subscription;
+    int executionCount;
+    readonly object gate = new();
 
     readonly Func<TInput, TOutput>? convert;     // for sync
     SingleAssignmentSubject<TInput>? asyncInput; // for async
 
     public event EventHandler? CanExecuteChanged;
 
+    public ReactiveProperty<bool> IsExecuting { get; } = new();
+
     public ReactiveCommand(Func<TInput, TOutput> convert)
     {
         this.list = new FreeListCore<Subscription>(this);
@@ -185,7 +244,7 @@ public ReactiveCommand(Func<TInput, CancellationToken, ValueTask<TOutput>> conve
         this.list = new FreeListCore<Subscription>(this);
         this.canExecute = true;
         this.asyncInput = new SingleAssignmentSubject<TInput>();
-        this.subscription = asyncInput.SelectAwait(convertAsync, awaitOperation, configureAwait, cancelOnCompleted, maxSequential).Subscribe(this, static (x, state) =>
+        this.subscription = asyncInput.SelectAwait((Command: this, ConvertAsync: convertAsync), static (value, state, cancellationToken) => state.Command.HandleAsyncExecution(value, state.ConvertAsync, cancellationToken), awaitOperation, configureAwait, cancelOnCompleted, maxSequential).Subscribe(this, static (x, state) =>
         {
             if (state.completeState.IsCompleted) return;
 
@@ -280,6 +339,56 @@ public void Execute(TInput parameter)
         }
     }
 
+    private void HandleExecution(TInput value, Action<TInput> execute)
+    {
+        try
+        {
+            lock (gate)
+            {
+                executionCount++;
+                IsExecuting.Value = executionCount > 0;
+            }
+
+            execute(value);
+        }
+        finally
+        {
+            lock (gate)
+            {
+                executionCount--;
+                if (executionCount == 0)
+                {
+                    IsExecuting.Value = false;
+                }
+            }
+        }
+    }
+
+    private async ValueTask<TOutput> HandleAsyncExecution(TInput value, Func<TInput, CancellationToken, ValueTask<TOutput>> executeAsync, CancellationToken cancellationToken)
+    {
+        try
+        {
+            lock (gate)
+            {
+                executionCount++;
+                IsExecuting.Value = executionCount > 0;
+            }
+
+            return await executeAsync(value, cancellationToken);
+        }
+        finally
+        {
+            lock (gate)
+            {
+                executionCount--;
+                if (executionCount == 0)
+                {
+                    IsExecuting.Value = false;
+                }
+            }
+        }
+    }
+
     protected override IDisposable SubscribeCore(Observer<TOutput> observer)
     {
         var result = completeState.TryGetResult();
@@ -420,7 +529,7 @@ public static ReactiveCommand<T> ToReactiveCommand<T>(
     {
         var command = new ReactiveCommand<T>(canExecuteSource, initialCanExecute);
 
-        var subscription = command.SubscribeAwait(async (x, ct) => await executeAsync(x, ct), awaitOperation, configureAwait, cancelOnCompleted, maxSequential);
+        var subscription = command.SubscribeAwait(executeAsync, static async (x, func, ct) => await func(x, ct), awaitOperation, configureAwait, cancelOnCompleted, maxSequential);
         command.CombineSubscription(subscription);
 
         return command;