Remove blocking get from donation jobs.

app/src/main/java/org/thoughtcrime/securesms/badges/gifts/flow/GiftFlowRepository.kt

@@ -18,8 +18,11 @@ import java.util.Locale
 class GiftFlowRepository {
 
   fun getGiftBadge(): Single<Pair<Long, Badge>> {
-    return ApplicationDependencies.getDonationsService()
+    return Single
+      .fromCallable {
+        ApplicationDependencies.getDonationsService()
           .getGiftBadges(Locale.getDefault())
+      }
       .flatMap(ServiceResponse<Map<Long, SignalServiceProfile.Badge>>::flattenResult)
       .map { gifts -> gifts.map { it.key to Badges.fromServiceBadge(it.value) } }
       .map { it.first() }
@@ -27,8 +30,11 @@ class GiftFlowRepository {
   }
 
   fun getGiftPricing(): Single<Map<Currency, FiatMoney>> {
-    return ApplicationDependencies.getDonationsService()
+    return Single
+      .fromCallable {
+        ApplicationDependencies.getDonationsService()
           .giftAmount
+      }
       .subscribeOn(Schedulers.io())
       .flatMap { it.flattenResult() }
       .map { result ->

app/src/main/java/org/thoughtcrime/securesms/badges/gifts/viewgift/ViewGiftRepository.kt

@@ -19,9 +19,12 @@ import java.util.Locale
 class ViewGiftRepository {
   fun getBadge(giftBadge: GiftBadge): Single<Badge> {
     val presentation = ReceiptCredentialPresentation(giftBadge.redemptionToken.toByteArray())
-    return ApplicationDependencies
+    return Single
+      .fromCallable {
+        ApplicationDependencies
           .getDonationsService()
           .getGiftBadge(Locale.getDefault(), presentation.receiptLevel)
+      }
       .flatMap { it.flattenResult() }
       .map { Badges.fromServiceBadge(it) }
       .subscribeOn(Schedulers.io())

app/src/main/java/org/thoughtcrime/securesms/components/settings/app/internal/donor/DonorErrorConfigurationViewModel.kt

@@ -26,20 +26,29 @@ class DonorErrorConfigurationViewModel : ViewModel() {
   val state: Flowable<DonorErrorConfigurationState> = store.stateFlowable
 
   init {
-    val giftBadges: Single<List<Badge>> = ApplicationDependencies.getDonationsService()
+    val giftBadges: Single<List<Badge>> = Single
+      .fromCallable {
+        ApplicationDependencies.getDonationsService()
           .getGiftBadges(Locale.getDefault())
+      }
       .flatMap { it.flattenResult() }
       .map { results -> results.values.map { Badges.fromServiceBadge(it) } }
       .subscribeOn(Schedulers.io())
 
-    val boostBadges: Single<List<Badge>> = ApplicationDependencies.getDonationsService()
+    val boostBadges: Single<List<Badge>> = Single
+      .fromCallable {
+        ApplicationDependencies.getDonationsService()
           .getBoostBadge(Locale.getDefault())
+      }
       .flatMap { it.flattenResult() }
       .map { listOf(Badges.fromServiceBadge(it)) }
       .subscribeOn(Schedulers.io())
 
-    val subscriptionBadges: Single<List<Badge>> = ApplicationDependencies.getDonationsService()
+    val subscriptionBadges: Single<List<Badge>> = Single
+      .fromCallable {
+        ApplicationDependencies.getDonationsService()
           .getSubscriptionLevels(Locale.getDefault())
+      }
       .flatMap { it.flattenResult() }
       .map { levels -> levels.levels.values.map { Badges.fromServiceBadge(it.badge) } }
       .subscribeOn(Schedulers.io())

app/src/main/java/org/thoughtcrime/securesms/components/settings/app/subscription/DonationPaymentRepository.kt

@@ -170,8 +170,11 @@ class DonationPaymentRepository(activity: Activity) : StripeApi.PaymentIntentFet
   fun cancelActiveSubscription(): Completable {
     Log.d(TAG, "Canceling active subscription...", true)
     val localSubscriber = SignalStore.donationsValues().requireSubscriber()
-    return ApplicationDependencies.getDonationsService()
+    return Single
+      .fromCallable {
+        ApplicationDependencies.getDonationsService()
           .cancelSubscription(localSubscriber.subscriberId)
+      }
       .subscribeOn(Schedulers.io())
       .flatMap(ServiceResponse<EmptyResponse>::flattenResult)
       .ignoreElement()
@@ -181,9 +184,12 @@ class DonationPaymentRepository(activity: Activity) : StripeApi.PaymentIntentFet
   fun ensureSubscriberId(): Completable {
     Log.d(TAG, "Ensuring SubscriberId exists on Signal service...", true)
     val subscriberId = SignalStore.donationsValues().getSubscriber()?.subscriberId ?: SubscriberId.generate()
-    return ApplicationDependencies
+    return Single
+      .fromCallable {
+        ApplicationDependencies
           .getDonationsService()
           .putSubscription(subscriberId)
+      }
       .subscribeOn(Schedulers.io())
       .flatMap(ServiceResponse<EmptyResponse>::flattenResult).ignoreElement()
       .doOnComplete {
@@ -268,13 +274,17 @@ class DonationPaymentRepository(activity: Activity) : StripeApi.PaymentIntentFet
         val subscriber = SignalStore.donationsValues().requireSubscriber()
 
         Log.d(TAG, "Attempting to set user subscription level to $subscriptionLevel", true)
+        Single
+          .fromCallable {
             ApplicationDependencies.getDonationsService().updateSubscriptionLevel(
               subscriber.subscriberId,
               subscriptionLevel,
               subscriber.currencyCode,
               levelUpdateOperation.idempotencyKey.serialize(),
               SubscriptionReceiptRequestResponseJob.MUTEX
-        ).flatMapCompletable {
+            )
+          }
+          .flatMapCompletable {
             if (it.status == 200 || it.status == 204) {
               Log.d(TAG, "Successfully set user subscription to level $subscriptionLevel with response code ${it.status}", true)
               SignalStore.donationsValues().updateLocalStateForLocalSubscribe()
@@ -356,9 +366,12 @@ class DonationPaymentRepository(activity: Activity) : StripeApi.PaymentIntentFet
 
   override fun fetchPaymentIntent(price: FiatMoney, level: Long): Single<StripeApi.PaymentIntent> {
     Log.d(TAG, "Fetching payment intent from Signal service for $price... (Locale.US minimum precision: ${price.minimumUnitPrecisionString})")
-    return ApplicationDependencies
+    return Single
+      .fromCallable {
+        ApplicationDependencies
           .getDonationsService()
           .createDonationIntentWithAmount(price.minimumUnitPrecisionString, price.currency.currencyCode, level)
+      }
       .flatMap(ServiceResponse<SubscriptionClientSecret>::flattenResult)
       .map {
         StripeApi.PaymentIntent(it.id, it.clientSecret)
@@ -370,7 +383,13 @@ class DonationPaymentRepository(activity: Activity) : StripeApi.PaymentIntentFet
   override fun fetchSetupIntent(): Single<StripeApi.SetupIntent> {
     Log.d(TAG, "Fetching setup intent from Signal service...")
     return Single.fromCallable { SignalStore.donationsValues().requireSubscriber() }
-      .flatMap { ApplicationDependencies.getDonationsService().createSubscriptionPaymentMethod(it.subscriberId) }
+      .flatMap {
+        Single.fromCallable {
+          ApplicationDependencies
+            .getDonationsService()
+            .createSubscriptionPaymentMethod(it.subscriberId)
+        }
+      }
       .flatMap(ServiceResponse<SubscriptionClientSecret>::flattenResult)
       .map { StripeApi.SetupIntent(it.id, it.clientSecret) }
       .doOnSuccess {
@@ -383,7 +402,11 @@ class DonationPaymentRepository(activity: Activity) : StripeApi.PaymentIntentFet
     return Single.fromCallable {
       SignalStore.donationsValues().requireSubscriber()
     }.flatMap {
-      ApplicationDependencies.getDonationsService().setDefaultPaymentMethodId(it.subscriberId, paymentMethodId)
+      Single.fromCallable {
+        ApplicationDependencies
+          .getDonationsService()
+          .setDefaultPaymentMethodId(it.subscriberId, paymentMethodId)
+      }
     }.flatMap(ServiceResponse<EmptyResponse>::flattenResult).ignoreElement().doOnComplete {
       Log.d(TAG, "Set default payment method via Signal service!")
     }

app/src/main/java/org/thoughtcrime/securesms/components/settings/app/subscription/SubscriptionsRepository.kt

@@ -23,7 +23,7 @@ class SubscriptionsRepository(private val donationsService: DonationsService) {
   fun getActiveSubscription(): Single<ActiveSubscription> {
     val localSubscription = SignalStore.donationsValues().getSubscriber()
     return if (localSubscription != null) {
-      donationsService.getSubscription(localSubscription.subscriberId)
+      Single.fromCallable { donationsService.getSubscription(localSubscription.subscriberId) }
         .subscribeOn(Schedulers.io())
         .flatMap(ServiceResponse<ActiveSubscription>::flattenResult)
     } else {
@@ -31,7 +31,8 @@ class SubscriptionsRepository(private val donationsService: DonationsService) {
     }
   }
 
-  fun getSubscriptions(): Single<List<Subscription>> = donationsService.getSubscriptionLevels(Locale.getDefault())
+  fun getSubscriptions(): Single<List<Subscription>> = Single
+    .fromCallable { donationsService.getSubscriptionLevels(Locale.getDefault()) }
     .subscribeOn(Schedulers.io())
     .flatMap(ServiceResponse<SubscriptionLevels>::flattenResult)
     .map { subscriptionLevels ->

app/src/main/java/org/thoughtcrime/securesms/components/settings/app/subscription/boost/BoostRepository.kt

@@ -5,6 +5,7 @@ import io.reactivex.rxjava3.schedulers.Schedulers
 import org.signal.core.util.money.FiatMoney
 import org.thoughtcrime.securesms.badges.Badges
 import org.thoughtcrime.securesms.badges.models.Badge
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
 import org.thoughtcrime.securesms.util.PlatformCurrencyUtil
 import org.whispersystems.signalservice.api.profiles.SignalServiceProfile
 import org.whispersystems.signalservice.api.services.DonationsService
@@ -16,7 +17,7 @@ import java.util.Locale
 class BoostRepository(private val donationsService: DonationsService) {
 
   fun getBoosts(): Single<Map<Currency, List<Boost>>> {
-    return donationsService.boostAmounts
+    return Single.fromCallable { donationsService.boostAmounts }
       .subscribeOn(Schedulers.io())
       .flatMap(ServiceResponse<Map<String, List<BigDecimal>>>::flattenResult)
       .map { result ->
@@ -28,7 +29,11 @@ class BoostRepository(private val donationsService: DonationsService) {
   }
 
   fun getBoostBadge(): Single<Badge> {
-    return donationsService.getBoostBadge(Locale.getDefault())
+    return Single
+      .fromCallable {
+        ApplicationDependencies.getDonationsService()
+          .getBoostBadge(Locale.getDefault())
+      }
       .subscribeOn(Schedulers.io())
       .flatMap(ServiceResponse<SignalServiceProfile.Badge>::flattenResult)
       .map(Badges::fromServiceBadge)

app/src/main/java/org/thoughtcrime/securesms/components/settings/app/subscription/receipts/detail/DonationReceiptDetailRepository.kt

@@ -9,9 +9,12 @@ import java.util.Locale
 
 class DonationReceiptDetailRepository {
   fun getSubscriptionLevelName(subscriptionLevel: Int): Single<String> {
-    return ApplicationDependencies
+    return Single
+      .fromCallable {
+        ApplicationDependencies
           .getDonationsService()
           .getSubscriptionLevels(Locale.getDefault())
+      }
       .flatMap { it.flattenResult() }
       .map { it.levels[subscriptionLevel.toString()] ?: throw Exception("Subscription level $subscriptionLevel not found") }
       .map { it.name }

app/src/main/java/org/thoughtcrime/securesms/components/settings/app/subscription/receipts/list/DonationReceiptListRepository.kt

@@ -9,7 +9,11 @@ import java.util.Locale
 
 class DonationReceiptListRepository {
   fun getBadges(): Single<List<DonationReceiptBadge>> {
-    val boostBadges: Single<List<DonationReceiptBadge>> = ApplicationDependencies.getDonationsService().getBoostBadge(Locale.getDefault())
+    val boostBadges: Single<List<DonationReceiptBadge>> = Single
+      .fromCallable {
+        ApplicationDependencies.getDonationsService()
+          .getBoostBadge(Locale.getDefault())
+      }
       .map { response ->
         if (response.result.isPresent) {
           listOf(DonationReceiptBadge(DonationReceiptRecord.Type.BOOST, -1, Badges.fromServiceBadge(response.result.get())))
@@ -18,7 +22,8 @@ class DonationReceiptListRepository {
         }
       }
 
-    val subBadges: Single<List<DonationReceiptBadge>> = ApplicationDependencies.getDonationsService().getSubscriptionLevels(Locale.getDefault())
+    val subBadges: Single<List<DonationReceiptBadge>> = Single
+      .fromCallable { ApplicationDependencies.getDonationsService().getSubscriptionLevels(Locale.getDefault()) }
       .map { response ->
         if (response.result.isPresent) {
           response.result.get().levels.map {

app/src/main/java/org/thoughtcrime/securesms/delete/DeleteAccountRepository.java

@@ -52,8 +52,7 @@ class DeleteAccountRepository {
 
         Subscriber                     subscriber                 = SignalStore.donationsValues().requireSubscriber();
         ServiceResponse<EmptyResponse> cancelSubscriptionResponse = ApplicationDependencies.getDonationsService()
-                                                                                           .cancelSubscription(subscriber.getSubscriberId())
+                                                                                           .cancelSubscription(subscriber.getSubscriberId());
-                                                                                           .blockingGet();
 
         if (cancelSubscriptionResponse.getExecutionError().isPresent()) {
           Log.w(TAG, "deleteAccount: failed attempt to cancel subscription");

app/src/main/java/org/thoughtcrime/securesms/glide/GiftBadgeModel.kt

@@ -47,7 +47,7 @@ data class GiftBadgeModel(val giftBadge: GiftBadge) : Key {
     override fun loadData(priority: Priority, callback: DataFetcher.DataCallback<in InputStream>) {
       try {
         val receiptCredentialPresentation = ReceiptCredentialPresentation(giftBadge.giftBadge.redemptionToken.toByteArray())
-        val giftBadgeResponse = ApplicationDependencies.getDonationsService().getGiftBadge(Locale.getDefault(), receiptCredentialPresentation.receiptLevel).blockingGet()
+        val giftBadgeResponse = ApplicationDependencies.getDonationsService().getGiftBadge(Locale.getDefault(), receiptCredentialPresentation.receiptLevel)
         if (giftBadgeResponse.result.isPresent) {
           val badge = Badges.fromServiceBadge(giftBadgeResponse.result.get())
           okHttpStreamFetcher = OkHttpStreamFetcher(client, GlideUrl(badge.imageUrl.toString()))

app/src/main/java/org/thoughtcrime/securesms/jobs/BoostReceiptRequestResponseJob.java

@@ -153,8 +153,7 @@ public class BoostReceiptRequestResponseJob extends BaseJob {
 
     Log.d(TAG, "Submitting credential to server", true);
     ServiceResponse<ReceiptCredentialResponse> response = ApplicationDependencies.getDonationsService()
-                                                                                 .submitBoostReceiptCredentialRequest(paymentIntentId, requestContext.getRequest())
+                                                                                 .submitBoostReceiptCredentialRequestSync(paymentIntentId, requestContext.getRequest());
-                                                                                 .blockingGet();
 
     if (response.getApplicationError().isPresent()) {
       handleApplicationError(context, response, donationErrorSource);

app/src/main/java/org/thoughtcrime/securesms/jobs/DonationReceiptRedemptionJob.java

@@ -149,8 +149,7 @@ public class DonationReceiptRedemptionJob extends BaseJob {
     ServiceResponse<EmptyResponse> response = ApplicationDependencies.getDonationsService()
                                                                      .redeemReceipt(presentation,
                                                                                     SignalStore.donationsValues().getDisplayBadgesOnProfile(),
-                                                                                    makePrimary)
+                                                                                    makePrimary);
-                                                                     .blockingGet();
 
     if (response.getApplicationError().isPresent()) {
       if (response.getStatus() >= 500) {

app/src/main/java/org/thoughtcrime/securesms/jobs/RefreshOwnProfileJob.java

@@ -276,8 +276,7 @@ public class RefreshOwnProfileJob extends BaseJob {
         boolean isDueToPaymentFailure = false;
         if (subscriber != null) {
           ServiceResponse<ActiveSubscription> response = ApplicationDependencies.getDonationsService()
-                                                                                .getSubscription(subscriber.getSubscriberId())
+                                                                                .getSubscription(subscriber.getSubscriberId());
-                                                                                .blockingGet();
 
           if (response.getResult().isPresent()) {
             ActiveSubscription activeSubscription = response.getResult().get();

app/src/main/java/org/thoughtcrime/securesms/jobs/SubscriptionKeepAliveJob.java

@@ -86,15 +86,13 @@ public class SubscriptionKeepAliveJob extends BaseJob {
     }
 
     ServiceResponse<EmptyResponse> response = ApplicationDependencies.getDonationsService()
-                                                                     .putSubscription(subscriber.getSubscriberId())
+                                                                     .putSubscription(subscriber.getSubscriberId());
-                                                                     .blockingGet();
 
     verifyResponse(response);
     Log.i(TAG, "Successful call to PUT subscription ID", true);
 
     ServiceResponse<ActiveSubscription> activeSubscriptionResponse = ApplicationDependencies.getDonationsService()
-                                                                                            .getSubscription(subscriber.getSubscriberId())
+                                                                                            .getSubscription(subscriber.getSubscriberId());
-                                                                                            .blockingGet();
 
     verifyResponse(activeSubscriptionResponse);
     Log.i(TAG, "Successful call to GET active subscription", true);

app/src/main/java/org/thoughtcrime/securesms/jobs/SubscriptionReceiptRequestResponseJob.java

@@ -187,8 +187,7 @@ public class SubscriptionReceiptRequestResponseJob extends BaseJob {
 
     Log.d(TAG, "Submitting receipt credential request.");
     ServiceResponse<ReceiptCredentialResponse> response = ApplicationDependencies.getDonationsService()
-                                                                                 .submitReceiptCredentialRequest(subscriberId, requestContext.getRequest())
+                                                                                 .submitReceiptCredentialRequestSync(subscriberId, requestContext.getRequest());
-                                                                                 .blockingGet();
 
     if (response.getApplicationError().isPresent()) {
       handleApplicationError(response);
@@ -216,8 +215,7 @@ public class SubscriptionReceiptRequestResponseJob extends BaseJob {
 
   private @NonNull ActiveSubscription getLatestSubscriptionInformation() throws Exception {
     ServiceResponse<ActiveSubscription> activeSubscription = ApplicationDependencies.getDonationsService()
-                                                                                    .getSubscription(subscriberId)
+                                                                                    .getSubscription(subscriberId);
-                                                                                    .blockingGet();
 
     if (activeSubscription.getResult().isPresent()) {
       return activeSubscription.getResult().get();

libsignal/service/src/main/java/org/whispersystems/signalservice/api/services/DonationsService.java

@@ -20,7 +20,6 @@ import org.whispersystems.signalservice.internal.push.PushServiceSocket;
 
 import java.io.IOException;
 import java.math.BigDecimal;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -28,8 +27,6 @@ import java.util.Objects;
 import java.util.TreeMap;
 
 import io.reactivex.rxjava3.annotations.NonNull;
-import io.reactivex.rxjava3.core.Single;
-import io.reactivex.rxjava3.schedulers.Schedulers;
 
 /**
  * One-stop shop for Signal service calls related to donations.
@@ -62,15 +59,13 @@ public class DonationsService {
    * @param visible                       Whether the badge will be visible on the user's profile immediately after redemption
    * @param primary                       Whether the badge will be made primary immediately after redemption
    */
-  public Single<ServiceResponse<EmptyResponse>> redeemReceipt(ReceiptCredentialPresentation receiptCredentialPresentation, boolean visible, boolean primary) {
+  public ServiceResponse<EmptyResponse> redeemReceipt(ReceiptCredentialPresentation receiptCredentialPresentation, boolean visible, boolean primary) {
-    return Single.fromCallable(() -> {
     try {
       pushServiceSocket.redeemDonationReceipt(receiptCredentialPresentation, visible, primary);
       return ServiceResponse.forResult(EmptyResponse.INSTANCE, 200, null);
     } catch (Exception e) {
       return ServiceResponse.<EmptyResponse>forUnknownError(e);
     }
-    }).subscribeOn(Schedulers.io());
   }
 
   /**
@@ -80,8 +75,8 @@ public class DonationsService {
    * @param currencyCode  The currency code for the amount
    * @return              A ServiceResponse containing a DonationIntentResult with details given to us by the payment gateway.
    */
-  public Single<ServiceResponse<SubscriptionClientSecret>> createDonationIntentWithAmount(String amount, String currencyCode, long level) {
+  public ServiceResponse<SubscriptionClientSecret> createDonationIntentWithAmount(String amount, String currencyCode, long level) {
-    return createServiceResponse(() -> new Pair<>(pushServiceSocket.createBoostPaymentMethod(currencyCode, Long.parseLong(amount), level), 200));
+    return wrapInServiceResponse(() -> new Pair<>(pushServiceSocket.createBoostPaymentMethod(currencyCode, Long.parseLong(amount), level), 200));
   }
 
   /**
@@ -91,36 +86,36 @@ public class DonationsService {
    * @param paymentIntentId          PaymentIntent ID from a boost donation intent response.
    * @param receiptCredentialRequest Client-generated request token
    */
-  public Single<ServiceResponse<ReceiptCredentialResponse>> submitBoostReceiptCredentialRequest(String paymentIntentId, ReceiptCredentialRequest receiptCredentialRequest) {
+  public ServiceResponse<ReceiptCredentialResponse> submitBoostReceiptCredentialRequestSync(String paymentIntentId, ReceiptCredentialRequest receiptCredentialRequest) {
-    return createServiceResponse(() -> new Pair<>(pushServiceSocket.submitBoostReceiptCredentials(paymentIntentId, receiptCredentialRequest), 200));
+    return wrapInServiceResponse(() -> new Pair<>(pushServiceSocket.submitBoostReceiptCredentials(paymentIntentId, receiptCredentialRequest), 200));
   }
 
   /**
    * @return The suggested amounts for Signal Boost
    */
-  public Single<ServiceResponse<Map<String, List<BigDecimal>>>> getBoostAmounts() {
+  public ServiceResponse<Map<String, List<BigDecimal>>> getBoostAmounts() {
-    return createServiceResponse(() -> new Pair<>(pushServiceSocket.getBoostAmounts(), 200));
+    return wrapInServiceResponse(() -> new Pair<>(pushServiceSocket.getBoostAmounts(), 200));
   }
 
   /**
    * @return The badge configuration for signal boost. Expect for right now only a single level numbered 1.
    */
-  public Single<ServiceResponse<SignalServiceProfile.Badge>> getBoostBadge(Locale locale) {
+  public ServiceResponse<SignalServiceProfile.Badge> getBoostBadge(Locale locale) {
-    return createServiceResponse(() -> new Pair<>(pushServiceSocket.getBoostLevels(locale).getLevels().get(SubscriptionLevels.BOOST_LEVEL).getBadge(), 200));
+    return wrapInServiceResponse(() -> new Pair<>(pushServiceSocket.getBoostLevels(locale).getLevels().get(SubscriptionLevels.BOOST_LEVEL).getBadge(), 200));
   }
 
   /**
    * @return A specific gift badge, by level.
    */
-  public Single<ServiceResponse<SignalServiceProfile.Badge>> getGiftBadge(Locale locale, long level) {
+  public ServiceResponse<SignalServiceProfile.Badge> getGiftBadge(Locale locale, long level) {
-    return createServiceResponse(() -> new Pair<>(pushServiceSocket.getBoostLevels(locale).getLevels().get(String.valueOf(level)).getBadge(), 200));
+    return wrapInServiceResponse(() -> new Pair<>(pushServiceSocket.getBoostLevels(locale).getLevels().get(String.valueOf(level)).getBadge(), 200));
   }
 
   /**
    * @return All gift badges the server currently has available.
    */
-  public Single<ServiceResponse<Map<Long, SignalServiceProfile.Badge>>> getGiftBadges(Locale locale) {
+  public ServiceResponse<Map<Long, SignalServiceProfile.Badge>> getGiftBadges(Locale locale) {
-    return createServiceResponse(() -> {
+    return wrapInServiceResponse(() -> {
       Map<String, SubscriptionLevels.Level> levels = pushServiceSocket.getBoostLevels(locale).getLevels();
       Map<Long, SignalServiceProfile.Badge> badges = new TreeMap<>();
 
@@ -141,15 +136,15 @@ public class DonationsService {
   /**
    * Returns the amounts for the gift badge.
    */
-  public Single<ServiceResponse<Map<String, BigDecimal>>> getGiftAmount() {
+  public ServiceResponse<Map<String, BigDecimal>> getGiftAmount() {
-    return createServiceResponse(() -> new Pair<>(pushServiceSocket.getGiftAmount(), 200));
+    return wrapInServiceResponse(() -> new Pair<>(pushServiceSocket.getGiftAmount(), 200));
   }
 
   /**
    * Returns the subscription levels that are available for the client to choose from along with currencies and current prices
    */
-  public Single<ServiceResponse<SubscriptionLevels>> getSubscriptionLevels(Locale locale) {
+  public ServiceResponse<SubscriptionLevels> getSubscriptionLevels(Locale locale) {
-    return createServiceResponse(() -> new Pair<>(pushServiceSocket.getSubscriptionLevels(locale), 200));
+    return wrapInServiceResponse(() -> new Pair<>(pushServiceSocket.getSubscriptionLevels(locale), 200));
   }
 
   /**
@@ -165,13 +160,13 @@ public class DonationsService {
    * @param idempotencyKey url-safe-base64-encoded random 16-byte value (see description)
    * @param mutex          A mutex to lock on to avoid a situation where this subscription update happens *as* we are trying to get a credential receipt.
    */
-  public Single<ServiceResponse<EmptyResponse>> updateSubscriptionLevel(SubscriberId subscriberId,
+  public ServiceResponse<EmptyResponse> updateSubscriptionLevel(SubscriberId subscriberId,
                                                                 String level,
                                                                 String currencyCode,
                                                                 String idempotencyKey,
                                                                 Object mutex
   ) {
-    return createServiceResponse(() -> {
+    return wrapInServiceResponse(() -> {
       synchronized(mutex) {
         pushServiceSocket.updateSubscriptionLevel(subscriberId.serialize(), level, currencyCode, idempotencyKey);
       }
@@ -180,10 +175,10 @@ public class DonationsService {
   }
 
   /**
-   * Returns information about the current subscription if one exists.
+   * Synchronously returns information about the current subscription if one exists.
    */
-  public Single<ServiceResponse<ActiveSubscription>> getSubscription(SubscriberId subscriberId) {
+  public ServiceResponse<ActiveSubscription> getSubscription(SubscriberId subscriberId) {
-    return createServiceResponse(() -> {
+    return wrapInServiceResponse(() -> {
       ActiveSubscription response = pushServiceSocket.getSubscription(subscriberId.serialize());
       return new Pair<>(response, 200);
     });
@@ -199,8 +194,8 @@ public class DonationsService {
    *
    * @param subscriberId  The subscriber ID for the user polling their subscription
    */
-  public Single<ServiceResponse<EmptyResponse>> putSubscription(SubscriberId subscriberId) {
+  public ServiceResponse<EmptyResponse> putSubscription(SubscriberId subscriberId) {
-    return createServiceResponse(() -> {
+    return wrapInServiceResponse(() -> {
       pushServiceSocket.putSubscription(subscriberId.serialize());
       return new Pair<>(EmptyResponse.INSTANCE, 200);
     });
@@ -211,15 +206,15 @@ public class DonationsService {
    *
    * @param subscriberId  The subscriber ID for the user cancelling their subscription
    */
-  public Single<ServiceResponse<EmptyResponse>> cancelSubscription(SubscriberId subscriberId) {
+  public ServiceResponse<EmptyResponse> cancelSubscription(SubscriberId subscriberId) {
-    return createServiceResponse(() -> {
+    return wrapInServiceResponse(() -> {
       pushServiceSocket.deleteSubscription(subscriberId.serialize());
       return new Pair<>(EmptyResponse.INSTANCE, 200);
     });
   }
 
-  public Single<ServiceResponse<EmptyResponse>> setDefaultPaymentMethodId(SubscriberId subscriberId, String paymentMethodId) {
+  public ServiceResponse<EmptyResponse> setDefaultPaymentMethodId(SubscriberId subscriberId, String paymentMethodId) {
-    return createServiceResponse(() -> {
+    return wrapInServiceResponse(() -> {
       pushServiceSocket.setDefaultSubscriptionPaymentMethod(subscriberId.serialize(), paymentMethodId);
       return new Pair<>(EmptyResponse.INSTANCE, 200);
     });
@@ -231,22 +226,21 @@ public class DonationsService {
    * @return              Client secret for a SetupIntent. It should not be used with the PaymentIntent stripe APIs
    *                      but instead with the SetupIntent stripe APIs.
    */
-  public Single<ServiceResponse<SubscriptionClientSecret>> createSubscriptionPaymentMethod(SubscriberId subscriberId) {
+  public ServiceResponse<SubscriptionClientSecret> createSubscriptionPaymentMethod(SubscriberId subscriberId) {
-    return createServiceResponse(() -> {
+    return wrapInServiceResponse(() -> {
       SubscriptionClientSecret clientSecret = pushServiceSocket.createSubscriptionPaymentMethod(subscriberId.serialize());
       return new Pair<>(clientSecret, 200);
     });
   }
 
-  public Single<ServiceResponse<ReceiptCredentialResponse>> submitReceiptCredentialRequest(SubscriberId subscriberId, ReceiptCredentialRequest receiptCredentialRequest) {
+  public ServiceResponse<ReceiptCredentialResponse> submitReceiptCredentialRequestSync(SubscriberId subscriberId, ReceiptCredentialRequest receiptCredentialRequest) {
-    return createServiceResponse(() -> {
+    return wrapInServiceResponse(() -> {
       ReceiptCredentialResponse response = pushServiceSocket.submitReceiptCredentials(subscriberId.serialize(), receiptCredentialRequest);
       return new Pair<>(response, 200);
     });
   }
 
-  private <T> Single<ServiceResponse<T>> createServiceResponse(Producer<T> producer) {
+  private <T> ServiceResponse<T> wrapInServiceResponse(Producer<T> producer) {
-    return Single.fromCallable(() -> {
     try {
       Pair<T, Integer> responseAndCode = producer.produce();
       return ServiceResponse.forResult(responseAndCode.first(), responseAndCode.second(), null);
@@ -257,7 +251,6 @@ public class DonationsService {
       Log.w(TAG, "An unknown error occurred.", e);
       return ServiceResponse.forUnknownError(e);
     }
-    });
   }
 
   private interface Producer<T> {

libsignal/service/src/test/java/org/whispersystems/signalservice/api/services/DonationsServiceTest.java

@@ -1,6 +1,5 @@
 package org.whispersystems.signalservice.api.services;
 
-import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -11,26 +10,18 @@ import org.whispersystems.signalservice.api.subscriptions.SubscriberId;
 import org.whispersystems.signalservice.internal.ServiceResponse;
 import org.whispersystems.signalservice.internal.push.PushServiceSocket;
 
-import io.reactivex.rxjava3.observers.TestObserver;
+import static org.junit.Assert.assertEquals;
-import io.reactivex.rxjava3.plugins.RxJavaPlugins;
+import static org.junit.Assert.assertFalse;
-import io.reactivex.rxjava3.schedulers.TestScheduler;
+import static org.junit.Assert.assertTrue;
-
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 @RunWith(JUnit4.class)
 public class DonationsServiceTest {
 
-  private static final TestScheduler TEST_SCHEDULER = new TestScheduler();
-
   private final PushServiceSocket pushServiceSocket = Mockito.mock(PushServiceSocket.class);
   private final DonationsService  testSubject       = new DonationsService(pushServiceSocket);
 
-  @BeforeClass
-  public static void setUpClass() {
-    RxJavaPlugins.setIoSchedulerHandler(scheduler -> TEST_SCHEDULER);
-  }
-
   @Test
   public void givenASubscriberId_whenIGetASuccessfulResponse_thenItIsMappedWithTheCorrectStatusCodeAndNonEmptyObject() throws Exception {
     // GIVEN
@@ -39,12 +30,12 @@ public class DonationsServiceTest {
         .thenReturn(getActiveSubscription());
 
     // WHEN
-    TestObserver<ServiceResponse<ActiveSubscription>> testObserver = testSubject.getSubscription(subscriberId).test();
+    ServiceResponse<ActiveSubscription> response = testSubject.getSubscription(subscriberId);
 
     // THEN
-    TEST_SCHEDULER.triggerActions();
     verify(pushServiceSocket).getSubscription(subscriberId.serialize());
-    testObserver.assertComplete().assertValue(value -> value.getStatus() == 200 && value.getResult().isPresent());
+    assertEquals(200, response.getStatus());
+    assertTrue(response.getResult().isPresent());
   }
 
   @Test
@@ -55,12 +46,12 @@ public class DonationsServiceTest {
         .thenThrow(new NonSuccessfulResponseCodeException(403));
 
     // WHEN
-    TestObserver<ServiceResponse<ActiveSubscription>> testObserver = testSubject.getSubscription(subscriberId).test();
+    ServiceResponse<ActiveSubscription> response = testSubject.getSubscription(subscriberId);
 
     // THEN
-    TEST_SCHEDULER.triggerActions();
     verify(pushServiceSocket).getSubscription(subscriberId.serialize());
-    testObserver.assertComplete().assertValue(value -> value.getStatus() == 403 && !value.getResult().isPresent());
+    assertEquals(403, response.getStatus());
+    assertFalse(response.getResult().isPresent());
   }
 
   private ActiveSubscription getActiveSubscription() {