io.reactivex.Single#fromCallable ( )源码实例Demo

@Override
public Single<TokenTicker> fetchTicker(Wallet wallet, Token token) {
    return Single.fromCallable(() -> {
        TokenTicker tokenTicker = null;
        try (Realm realm = realmManager.getRealmInstance(wallet))
        {
            RealmTokenTicker rawItem = realm.where(RealmTokenTicker.class)
                    .equalTo("contract", token.getAddress() + "-" + token.tokenInfo.chainId)
                    .findFirst();

            tokenTicker = convertRealmTicker(rawItem);
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        return tokenTicker == null
                ? new TokenTicker()
                : tokenTicker;
    });
}
 

/**
 * Fetches a single row from a database and maps it to and object of type {@link T},
 * non-blocking operation.
 *
 * @return a {@link Single<T>} where an object of type {@link T} mapped from a database
 * record is passed as the parameter to
 * {@link io.reactivex.observers.DisposableSingleObserver#onSuccess(Object)}
 */
@Override
public Single<T> asSingle() {
    return Single.fromCallable(new Callable<T>() {
        @Override
        public T call() {
            return executeBlocking();
        }
    });
}
 

@Override
public Single<Integer> getWaitingResultTimeout() {
    return Single.fromCallable(() ->
            context.getSharedPreferences(CONFIG_FILE, Context.MODE_PRIVATE)
                    .getInt(KEY_WAITING_RESULT_TIMEOUT, 120)
    );
}
 

@Override
public <E extends T> Single<E> insert(final E entity) {
    return Single.fromCallable(new Callable<E>() {
        @Override
        public E call() throws Exception {
            return delegate.insert(entity);
        }
    });
}
 

/**
 * Creates a {@link Single} that when subscribed to executes this compiled
 * delete statement against a database and emits nr of deleted records to downstream
 * only once.
 *
 * @return Deferred {@link Single} that when subscribed to executes the statement and emits
 * its result to downstream
 */
@NonNull
@CheckResult
public Single<Integer> observe() {
  return Single.fromCallable(new Callable<Integer>() {
    @Override
    public Integer call() throws Exception {
      return execute();
    }
  });
}
 

public Single<List<AppVersion>> application(String name) {
    return Single.fromCallable(() -> Arrays.asList(
            new AppVersion("1.0.0", "38", "22/11/2017 23h59", ""),
            new AppVersion("1.0.0", "38", "22/11/2017 23h59", ""),
            new AppVersion("1.0.0", "38", "22/11/2017 23h59", ""),
            new AppVersion("1.0.0", "38", "22/11/2017 23h59", ""),
            new AppVersion("1.0.0", "38", "22/11/2017 23h59", ""),
            new AppVersion("1.0.0", "38", "22/11/2017 23h59", ""),
            new AppVersion("1.0.0", "38", "22/11/2017 23h59", ""),
            new AppVersion("1.0.0", "38", "22/11/2017 23h59", ""),
            new AppVersion("1.0.0", "38", "22/11/2017 23h59", "")
    ));
}
 

private Single<String> storeUnconfirmedTransaction(Wallet from, String txHash, String toAddress, BigInteger value, BigInteger gasPrice, int chainId, String data)
{
	return Single.fromCallable(() -> {

		Transaction newTx = new Transaction(txHash, "0", "0", System.currentTimeMillis()/1000, 0, from.address, toAddress, value.toString(10), "0", gasPrice.toString(10), data,
				"0", chainId, "");
		inDiskCache.putTransaction(from, newTx);

		return txHash;
	});
}
 

private Single<int[]> generateTicketArray(String indices, Ticket ticket)
{
    return Single.fromCallable(() -> {
        List<Integer> ticketIndices = ticket.stringIntsToIntegerList(indices);
        int[] indicesArray = new int[ticketIndices.size()];
        for (int i = 0; i < ticketIndices.size(); i++) indicesArray[i] = ticketIndices.get(i);
        return indicesArray;
    });
}
 

@Override
public Single<ContractType> queryInterfaceSpec(String address, TokenInfo tokenInfo)
{
	NetworkInfo networkInfo = networkRepository.getNetworkByChain(tokenInfo.chainId);
	ContractType checked = TokensService.checkInterfaceSpec(tokenInfo.chainId, tokenInfo.address);
	if (tokenInfo.name == null && tokenInfo.symbol == null)
	{
		return Single.fromCallable(() -> ContractType.NOT_SET);
	}
	else if (checked != null && checked != ContractType.NOT_SET && checked != ContractType.OTHER)
	{
		return Single.fromCallable(() -> checked);
	}
	else return blockExplorerClient.checkConstructorArgs(networkInfo, address);
}
 

@Override
public <K, E extends T> Single<K> insert(final E entity, final Class<K> keyClass) {
    return Single.fromCallable(new Callable<K>() {
        @Override
        public K call() throws Exception {
            return delegate.insert(entity, keyClass);
        }
    });
}
 

private Single<Boolean> closeListener()
{
    return Single.fromCallable(() -> {
        try {
            memPoolSubscription.dispose();
            return true;
        } catch (NetworkOnMainThreadException th) {
            // Ignore all errors, it's not important source.
            return false;
        }
    });
}
 

private Single<byte[]> encodeTransaction(byte[] signatureBytes, RawTransaction rtx)
{
	return Single.fromCallable(() -> {
		Sign.SignatureData sigData = sigFromByteArray(signatureBytes);
		if (sigData == null) return FAILED_SIGNATURE.getBytes();
		return encode(rtx, sigData);
	});
}
 

public Single<Boolean> getWalletBackupWarning(String walletAddr)
{
    return Single.fromCallable(() -> {
        long backupTime = getKeyBackupTime(walletAddr);
        long warningTime = getWalletWarningTime(walletAddr);
        return requiresBackup(backupTime, warningTime);
    });
}
 

@Override
public Single<Boolean> deleteFavoriteMovie(@NonNull final MovieData movie) {
    return Single.fromCallable(() -> {
        final FavoriteDao dao = favoriteDatabase.getDao();
        return dao.deleteFavorite(movie.getId()) == 1;
    });
}
 

public Single<List<Address>> fromLocationName(Locale locale, @NonNull String locationName, int maxResults) {
    return Single.fromCallable(() -> getGeocoder(locale).getFromLocationName(locationName, maxResults));
}
 

@Override
public Single<String> getUserName() {
    return Single.fromCallable(() -> userModule.authenticatedUser().blockingGet().user());
}
 

private Single<File> fetchXMLFromServer(String address)
{
    return Single.fromCallable(() -> {
        final File defaultReturn = new File("");
        if (address.equals("")) return defaultReturn;

        File result = getDownloadedXMLFile(address);

        //peek to see if this file exists
        long fileTime = 0;
        if (result != null && result.exists())
        {
            TokenDefinition td = getTokenDefinition(result);
            if (definitionIsOutOfDate(td))
            {
                removeFile(result.getAbsolutePath());
                assetChecked.put(address, 0L);
            }
            else
            {
                fileTime = result.lastModified();
            }
        }
        else
        {
            result = defaultReturn;
        }

        if (assetChecked.get(address) != null && (System.currentTimeMillis() > (assetChecked.get(address) + 1000L*60L*60L))) return result;

        SimpleDateFormat format = new SimpleDateFormat("EEE, d MMM yyyy HH:mm:ss 'GMT'", Locale.ENGLISH);
        format.setTimeZone(TimeZone.getTimeZone("UTC"));
        String dateFormat = format.format(new Date(fileTime));

        StringBuilder sb = new StringBuilder();
        sb.append(TOKENSCRIPT_REPO_SERVER);
        sb.append(TOKENSCRIPT_CURRENT_SCHEMA);
        sb.append("/");
        sb.append(address);

        //prepare Android headers
        PackageManager manager = context.getPackageManager();
        PackageInfo info = manager.getPackageInfo(
                context.getPackageName(), 0);
        String appVersion = info.versionName;
        String OSVersion = String.valueOf(Build.VERSION.RELEASE);

        okhttp3.Response response = null;

        try
        {
            Request request = new Request.Builder()
                    .url(sb.toString())
                    .get()
                    .addHeader("Accept", "text/xml; charset=UTF-8")
                    .addHeader("X-Client-Name", "AlphaWallet")
                    .addHeader("X-Client-Version", appVersion)
                    .addHeader("X-Platform-Name", "Android")
                    .addHeader("X-Platform-Version", OSVersion)
                    .addHeader("If-Modified-Since", dateFormat)
                    .build();

            response = okHttpClient.newCall(request).execute();

            switch (response.code())
            {
                case HttpURLConnection.HTTP_NOT_MODIFIED:
                    result = defaultReturn;
                    break;
                case HttpURLConnection.HTTP_OK:
                    String xmlBody = response.body().string();
                    result = storeFile(address, xmlBody);
                    break;
                default:
                    result = defaultReturn;
                    break;
            }
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            if (response != null) response.body().close();
        }

        assetChecked.put(address, System.currentTimeMillis());

        return result;
    });
}
 

@Override
public Single<List<TrackedEntityInstance>> get() {
    return Single.fromCallable(this::blockingGet);
}
 

private Single<Transaction[]> noTransactions()
{
	return Single.fromCallable(() -> new Transaction[0]);
}
 

public Single<List<TrackedEntityInstance>> getCall(final Collection<String> trackedEntityInstanceUids,
                                                   final String program) {
    return Single.fromCallable(() -> downloadAndPersistBlocking(trackedEntityInstanceUids, program));
}