In my recent articles I argued (among other things) that concurrency frameworks are neither required nor sufficient to write clean concurrent code. In other words, you can write clean concurrent code without concurrency frameworks, and if you do use them, it won’t automatically make your code better.

This idea turned out to be both interesting and controversial because it led to two distinct types of feedback. One group of readers got interested and asked how to implement more complex concurrent requirements without frameworks. Another group of readers expressed their deepest disagreement with my statements and some developers even argued that my ideas are harmful and dangerous.

In this post I’m going to show you how I write complex and clean concurrent code without frameworks. In my opinion, this demonstration should cover both the questions about more complex scenarios and the concerns about the feasibility of this approach.

Requirements

The example which I’ll show you in this article is based on a real-world scenario that I dealt with in client’s codebase last week. Obviously, I can’t share the details and I can’t show you client’s code, but, luckily, these aren’t required to demonstrate the fundamentals of concurrent code.

So, imagine that you have these requirements for a specific app’s feature:

1. Process and merge two groups of files into two new individual files
2. Compress the resulting files into zip archive
3. Upload zip archive to a server
4. Notify other interested components within the codebase about the result of flow’s execution

Since files and networking are involved, this flow shouldn’t be executed on UI thread. However, I do want to be able to call it from UI thread and I do want to get completion notifications on UI thread. Therefore, I’ll need to write concurrent code.

Concurrency Implementation

When I implemented the requirements, client’s codebase already used ThreadPoster library. Therefore, I used it for this new functionality as well. However, to make a point, in this article I’ll use bare Thread class for background work. To return execution from background to UI thread, I’ll use the standard Android’s Handler class.

Basic Implementation

The most basic implementation of the above flow can be implemented in the following manner:

public class UploadFilesUseCase extends BaseObservable<UploadFilesUseCase.Listener> {

    public interface Listener {
        void onFilesUploaded();
        void onFilesUploadFailed();
    }

    private final Handler uiHandler = new Handler(Looper.getMainLooper());

    public void uploadFiles() {
        new Thread(() -> uploadFilesSync()).start();
    }

    @WorkerThread
    private void uploadFilesSync() {
        File mergedA = processAndMergeFilesOfTypeA();
        File mergedB = processAndMergeFilesOfTypeB();
        File archive = compressMergedFiles(mergedA, mergedB);

        HttpManager.getInstance.uploadFiles(
                archive,
                new HttpRequestListener() {
                    @Override
                    public void onDone(int code, byte[] body) {
                        if (code / 100 == 2) {
                            notifySuccess();
                        } else {
                            notifyFailure();
                        }
                    }

                    @Override
                    public void onFailure() {
                        notifyFailure();
                    }
                }
        );
    }

    @WorkerThread
    private File processAndMergeFilesOfTypeA() { ... }

    @WorkerThread
    private File processAndMergeFilesOfTypeB() { ... }

    @WorkerThread
    private File compressMergedFiles(File fileA, File fileB) { ... }

    private void notifySuccess() {
        uiHandler.post(() -> {
            for (Listener listener : getListeners()) {
                listener.onFilesUploaded();
            }
        });
    }

    private void notifyFailure() {
        uiHandler.post(() -> {
            for (Listener listener : getListeners()) {
                listener.onFilesUploadFailed();
            }
        });
    }

}

If you don’t understand why this “use case” class is structured the way it is, then you can read this article about use cases.

I left the methods related to file manipulations blank. Their logic is not relevant to our discussion here.

You might be surprised by the fact that I use HttpManager as Singleton. Given that I always advocate against Singletons, that’s odd. Well, if I’d write this app from scratch, I wouldn’t do that. However, in this case, I had to work with legacy code and refactoring of networking layer wasn’t the goal of this project. Theoretically, I could inject HttpManager into constructor to make this class a bit cleaner. However, I knew that the codebase suffered from bad temporal coupling and I didn’t want taking on additional risk by (potentially) changing the initialization order.

On the other hand, this Singleton with its callbacks will allow us to make this example a bit more interesting later, so that’s a plus.

Error Handling

The above basic implementation accounts for networking errors, but network isn’t the only source of potential errors in this flow. File operations can fail too.

Therefore, let’s assume that file-related methods can throw checked OperationFailedException if anything goes wrong:

@WorkerThread
private File processAndMergeFilesOfTypeA() throws OperationFailedException { ... }

@WorkerThread
private File processAndMergeFilesOfTypeB() throws OperationFailedException { ... }

@WorkerThread
private File compressMergedFiles(File fileA, File fileB) throws OperationFailedException { ... }

Then I can add the following error handling:

@WorkerThread
private void uploadFilesSync() {
    File archive = null;
    try {
        File mergedA = processAndMergeFilesOfTypeA();
        File mergedB = processAndMergeFilesOfTypeB();
        archive = compressMergedFiles(mergedA, mergedB);
    } catch (OperationFailedException e) {
        notifyFailure();
        return;
    }

    HttpManager.getInstance.uploadFiles(
            archive,
            new HttpRequestListener() {
                @Override
                public void onDone(int code, byte[] body) {
                    if (code / 100 == 2) {
                        notifySuccess();
                    } else {
                        notifyFailure();
                    }
                }

                @Override
                public void onFailure() {
                    notifyFailure();
                }
            }
    );
}

This is very simple error handling, but you can implement much more complex schemes if needed.

For example, let’s say that in case of any error I want to retry this flow for a predefined number of times, before I declare it a “failure”. That would be surprisingly simple to achieve:

public void uploadFiles() {
    new Thread(() -> uploadFilesSync(0)).start();
}

@WorkerThread
private void uploadFilesSync(int retryCount) {
    File archive = null;
    try {
        File mergedA = processAndMergeFilesOfTypeA();
        File mergedB = processAndMergeFilesOfTypeB();
        archive = compressMergedFiles(mergedA, mergedB);
    } catch (OperationFailedException e) {
        retryOrFail(retryCount);
        return;
    }

    HttpManager.getInstance.uploadFiles(
            archive,
            new HttpRequestListener() {
                @Override
                public void onDone(int code, byte[] body) {
                    if (code / 100 == 2) {
                        notifySuccess();
                    } else {
                        retryOrFail(retryCount);
                    }
                }

                @Override
                public void onFailure() {
                    retryOrFail(retryCount);
                }
            }
    );
}

@WorkerThread
private void retryOrFail(int currentRetryCount) {
    if (currentRetryCount >= MAX_RETRIES - 1) {
        notifyFailure();
        return;
    }
    uploadFilesSync(currentRetryCount + 1);
}

As an exercise, I invite you to think how you’d implement exponential back-off for retry attempts. Hint: only retryOrFail method would be affected.

Using this approach you can implement clean error handling of any complexity.

Performance Optimization

The implementation that I have by now is great and will suffice in most situations. However, very rarely, you might need to optimize it. This was one of these rare cases.

See, the files in question are relatively big, so this flow took a considerable time. While it’s not the end of the world, client’s staff initiated the upload remotely and then waited to get the files. Given the fact that they’ll do it hundreds, or even thousands of times in the future, I decided that it’s worth complicating the implementation of this class to reduce the execution time.

What can be optimized here? Well, not that much because most of the steps in this flow are inter-dependent. However, the processing of two sets of files can be done concurrently. Therefore, I decided to process every set of files on a dedicated background thread.

The resulting implementation looks like this:

@WorkerThread
private void uploadFilesSync(int retryCount) {

    final AtomicReference<File> mergedA = new AtomicReference<>(null);
    final AtomicReference<File> mergedB = new AtomicReference<>(null);

    final CountDownLatch countDownLatch = new CountDownLatch(2);

    new Thread(() -> {
        try {
            mergedA.set(processAndMergeFilesOfTypeA());
        } catch (OperationFailedException e) {
            // log the exception
        } finally {
            countDownLatch.countDown();
        }
    }).start();

    new Thread(() -> {
        try {
            mergedB.set(processAndMergeFilesOfTypeB());
        } catch (OperationFailedException e) {
            // log the exception
        } finally {
            countDownLatch.countDown();
        }
    }).start();

    try {
        countDownLatch.await();
    } catch (InterruptedException e) {
        throw new RuntimeException("unexpected interrupt");
    }

    if (mergedA.get() == null || mergedB.get() == null) {
        retryOrFail(retryCount);
        return;
    }

    File archive;
    try {
        archive = compressMergedFiles(mergedA.get(), mergedB.get());
    } catch (OperationFailedException e) {
        retryOrFail(retryCount);
        return;
    }

    HttpManager.getInstance.uploadFiles(
            archive,
            new HttpRequestListener() {
                @Override
                public void onDone(int code, byte[] body) {
                    if (code / 100 == 2) {
                        notifySuccess();
                    } else {
                        retryOrFail(retryCount);
                    }
                }

                @Override
                public void onFailure() {
                    retryOrFail(retryCount);
                }
            }
    );
}

As you see, performance optimization using concurrency caused quite a jump in complexity of this class. That’s why I recommend avoiding optimization for the sake of optimization. Do that only if you have clear and compelling business case which justifies the trade-off.

Cleanup Of Temporary Files

The flow encapsulated in this use case class creates temporary files along the way. They aren’t needed after the flow completes and can even cause troubles. Therefore, I want to clean them up.

When should these files be deleted? When the flow either completes successfully or fails, of course. That’s right, but there is more. Since we added retries to this algorithm, I’d like to employ defensive programming and delete these files on each retry.

To simplify the implementation, I made sure that temp files are always created in one specific directory. This way the cleanup amounts to just deleting this directory each time.

That’s the result:

@WorkerThread
private void uploadFilesSync(int retryCount) {

    final AtomicReference<File> mergedA = new AtomicReference<>(null);
    final AtomicReference<File> mergedB = new AtomicReference<>(null);

    final CountDownLatch countDownLatch = new CountDownLatch(2);

    new Thread(() -> {
        try {
            mergedA.set(processAndMergeFilesOfTypeA());
        } catch (OperationFailedException e) {
            // log the exception
        } finally {
            countDownLatch.countDown();
        }
    }).start();

    new Thread(() -> {
        try {
            mergedB.set(processAndMergeFilesOfTypeB());
        } catch (OperationFailedException e) {
            // log the exception
        } finally {
            countDownLatch.countDown();
        }
    }).start();

    try {
        countDownLatch.await();
    } catch (InterruptedException e) {
        throw new RuntimeException("unexpected interrupt");
    }

    if (mergedA.get() == null || mergedB.get() == null) {
        retryOrFail(retryCount);
        return;
    }

    File archive;
    try {
        archive = compressMergedFiles(mergedA.get(), mergedB.get());
    } catch (OperationFailedException e) {
        retryOrFail(retryCount);
        return;
    }

    HttpManager.getInstance.uploadFiles(
            archive,
            new HttpRequestListener() {
                @Override
                public void onDone(int code, byte[] body) {
                    if (code / 100 == 2) {
                        notifySuccess();
                    } else {
                        retryOrFail(retryCount);
                    }
                }

                @Override
                public void onFailure() {
                    retryOrFail(retryCount);
                }
            }
    );

    deleteTempDir();
}

@WorkerThread
private void retryOrFail(int currentRetryCount) {
    deleteTempDir();
    if (currentRetryCount >= MAX_RETRIES - 1) {
        notifyFailure();
        return;
    }
    uploadFilesSync(currentRetryCount + 1);
}

Have you noticed the bug, though? If you haven’t, go on and look for it. This was a real bug that I planted in the original implementation and found during testing.

Hint: the uploaded file will always be 0 bytes in size.

The problem, of course, is that I start the upload of the archive and concurrently delete the temporary directory. Not a smart thing to do.

The fix is quite simple: just move the final call to deleteTempDir() into HttpRequestListener’s onDone method:

    HttpManager.getInstance.uploadFiles(
            archive,
            new HttpRequestListener() {
                @Override
                public void onDone(int code, byte[] body) {
                    if (code / 100 == 2) {
                        deleteTempDir();
                        notifySuccess();
                    } else {
                        retryOrFail(retryCount);
                    }
                }

                @Override
                public void onFailure() {
                    retryOrFail(retryCount);
                }
            }
    );

Now it should work as expected.

Synchronizing Asynchronous Code

One of the reasons for the bug in the previous section is that I mixed synchronous code with asynchronous callback. In essence, by using HttpManager inside this use case, I made the first step towards so-called callback hell. That bug was a direct consequence of the additional complexity.

As I already wrote, refactoring the networking layer of this app wasn’t on the table. So, looks like I had no other options left. Not exactly. I could still convert the entire flow to synchronous execution.

That’s how I could do that:

@WorkerThread
private void uploadFilesSync(int retryCount) {

    final AtomicReference<File> mergedA = new AtomicReference<>(null);
    final AtomicReference<File> mergedB = new AtomicReference<>(null);

    final CountDownLatch countDownLatch = new CountDownLatch(2);

    new Thread(() -> {
        try {
            mergedA.set(processAndMergeFilesOfTypeA());
        } catch (OperationFailedException e) {
            // log the exception
        } finally {
            countDownLatch.countDown();
        }
    }).start();

    new Thread(() -> {
        try {
            mergedB.set(processAndMergeFilesOfTypeB());
        } catch (OperationFailedException e) {
            // log the exception
        } finally {
            countDownLatch.countDown();
        }
    }).start();

    try {
        countDownLatch.await();
    } catch (InterruptedException e) {
        throw new RuntimeException("unexpected interrupt");
    }

    if (mergedA.get() == null || mergedB.get() == null) {
        retryOrFail(retryCount);
        return;
    }

    File archive;
    try {
        archive = compressMergedFiles(mergedA.get(), mergedB.get());
    } catch (OperationFailedException e) {
        retryOrFail(retryCount);
        return;
    }

   int responseCode = uploadFileToServer(archive);

    if (responseCode / 100 == 2) {
        deleteTempDir();
        notifySuccess();
    } else {
        retryOrFail(retryCount);
    }
}

@WorkerThread
private int uploadFileToServer(File archive) {
    final AtomicInteger responseCode = new AtomicInteger(0);
    final CountDownLatch countDownLatch = new CountDownLatch(1);

    HttpManager.getInstance.uploadFiles(
            archive,
            new HttpRequestListener() {
                @Override
                public void onDone(int code, byte[] body) {
                    responseCode.set(code);
                    countDownLatch.countDown();
                }

                @Override
                public void onFailure() {
                    responseCode.set(0);
                    countDownLatch.countDown();
                }
            }
    );

    try {
        countDownLatch.await();
    } catch (InterruptedException e) {
        throw new RuntimeException("unexpected interrupt");
    }

    return responseCode.get();
}

One exceptionally important point to note here is that even though the code became longer, it also became much simpler. Now the algorithm in uploadFilesSync() is crystal clear and its steps form a simple sequential structure. That’s a big win for long-term maintainability.

However, that’s not all. There are also major design benefits.

Note that the original implementation assumed (implicitly) that HttpManager will invoke the callback on a background thread. If it would use UI thread instead, the app would freeze for prolonged periods of time. The “synchronized” implementation makes no such assumption anymore, so it’s less coupled to the implementation of HttpManager.

In addition, note that uploadFileToServer() method doesn’t depend on anything inside this use case. Therefore, I can extract it into standalone class which will abstract out all the irrelevant details, including the fact that HttpManager is a Singleton. If I’d do the same for other methods inside HttpManager, I’d create convenient synchronous networking API inside the codebase. Furthermore, if I’d break the API of HttpManager into multiple narrowly focused classes, it would be highly beneficial if I’d need to actually refactor the networking layer in this app. That’s so-called Interface Segregation Principle in action.

So, implementation with more code, but much better in pretty much every aspect. I want you to remember this point because there is so much hype about “less lines of code” out there, which is absolutely meaningless metric on this scale. Especially in concurrent code.

Preventing Concurrent Invocations

After I implemented all the business requirements in this use case, I realized that there is one additional implicit technical requirement here: prevent multiple concurrent invocations of this flow. It doesn’t make any sense to upload the same data multiple times concurrently and could lead to serious bugs, including data corruption.

I meet this requirement quite often, so I already have a standard pattern to implement it. First, I’ll make UploadFilesUseCase extend BaseBusyObservable instead of just BaseObservable. This will add “busy” flag to the use case, as well as several convenience methods to handle transitions to and from “busy” state.

Once I have that, the implementation of this requirement amounts to proper management of “busy” flag and aborting use case invocations when it’s “busy”.

Final Implementation

The final implementation of this complex use case, excluding irrelevant details, will look like this:

public class UploadFilesUseCase extends BaseBusyObservable<UploadFilesUseCase.Listener> {

    public interface Listener {
        void onFilesUploaded();
        void onFilesUploadFailed();
    }

    private final static int MAX_RETRIES = 3;

    private final Handler uiHandler = new Handler(Looper.getMainLooper());

    public void uploadFiles() {
        if (!isFreeAndBecomeBusy()) {
            // log concurrent invocation attempt
            return;
        }
        new Thread(() -> uploadFilesSync(0)).start();
    }

    @WorkerThread
    private void uploadFilesSync(int retryCount) {
        final AtomicReference<File> mergedA = new AtomicReference<>(null);
        final AtomicReference<File> mergedB = new AtomicReference<>(null);

        final CountDownLatch countDownLatch = new CountDownLatch(2);

        new Thread(() -> {
            try {
                mergedA.set(processAndMergeFilesOfTypeA());
            } catch (OperationFailedException e) {
                // log the exception
            } finally {
                countDownLatch.countDown();
            }
        }).start();

        new Thread(() -> {
            try {
                mergedB.set(processAndMergeFilesOfTypeB());
            } catch (OperationFailedException e) {
                // log the exception
            } finally {
                countDownLatch.countDown();
            }
        }).start();

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException("unexpected interrupt");
        }

        if (mergedA.get() == null || mergedB.get() == null) {
            retryOrFail(retryCount);
            return;
        }

        File archive;
        try {
            archive = compressMergedFiles(mergedA.get(), mergedB.get());
        } catch (OperationFailedException e) {
            retryOrFail(retryCount);
            return;
        }

       int responseCode = uploadFileToServer(archive);

        if (responseCode / 100 == 2) {
            deleteTempDir();
            notifySuccess();
        } else {
            retryOrFail(retryCount);
        }
    }

    @WorkerThread
    private int uploadFileToServer(File archive) {
        final AtomicInteger responseCode = new AtomicInteger(0);
        final CountDownLatch countDownLatch = new CountDownLatch(1);

        HttpManager.getInstance.uploadFiles(
                archive,
                new HttpRequestListener() {
                    @Override
                    public void onDone(int code, byte[] body) {
                        responseCode.set(code);
                        countDownLatch.countDown();
                    }

                    @Override
                    public void onFailure() {
                        responseCode.set(0);
                        countDownLatch.countDown();
                    }
                }
        );

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException("unexpected interrupt");
        }

        return responseCode.get();
    }

    @WorkerThread
    private void retryOrFail(int currentRetryCount) {
        deleteTempDir();
        if (currentRetryCount >= MAX_RETRIES - 1) {
            notifyFailure();
            return;
        }
        uploadFilesSync(currentRetryCount + 1);
    }

    @WorkerThread
    private File processAndMergeFilesOfTypeA() throws OperationFailedException { ... }

    @WorkerThread
    private File processAndMergeFilesOfTypeB() throws OperationFailedException { ... }

    @WorkerThread
    private File compressMergedFiles(File fileA, File fileB) throws OperationFailedException { ... }

    @WorkerThread
    private void deleteTempDir() { ... }

    private void notifySuccess() {
        uiHandler.post(() -> {
            for (Listener listener : getListeners()) {
                listener.onFilesUploaded();
            }
            becomeNotBusy();
        });
    }

    private void notifyFailure() {
        uiHandler.post(() -> {
            for (Listener listener : getListeners()) {
                listener.onFilesUploadFailed();
            }
            becomeNotBusy();
        });
    }

}

Evidently, there is quite a bit of code in this class. Some of it might be extracted into collaborator objects, but I kept it grouped in a single class for the purpose of this article.

However, what’s important here is not the amount of code, but the fact that every developer who understands concurrency on JVM can maintain this class. Even though the requirements are complex, the code is very explicit about what’s going on and no knowledge of any framework is required. As you saw, even something as complex as exponential back-off can be added to this implementation by making small change in one single method.

This code would be good ten years ago (with some syntactic changes), it’s good today and it’s going to be good ten years from now. That’s what makes for maintainable code, not the number of lines of code.

Alternative Implementations Using Frameworks

I posted this article on Reddit to get a bit of critical feedback, and, boy, did I get it. There are some very interesting discussions there and other developers started to provide alternative implementations. I invite you to read that thread to see other opinions on this topic.

Currently, this gist is the best alternative implementation using RxJava. It uses Kotlin instead of Java, assumes different implementation of HttpManager and lacks logging, but, in principle, it should implement similar requirements. You can see that it’s much shorter than my original implementation and has less “moving parts” (at least on paper). However, even though I’m familiar with RxJava in general, I can’t verify the correctness of this solution from just reading the code. This has always been my biggest problem with this framework: you need to invest a lot of time into it to be able to use it, or even just read someone else’s code. There is no way to reason about RxJava code from the fundamental principles. Therefore, I always recommended against coupling your project to this third-party framework. In addition, RxJava is long past its popularity peak and will soon become legacy in Android.

The version which uses Coroutines is much more interesting in my opinion. It’s definitely much shorter than a solution that uses multithreading primitives and easier to understand. However, it’s not simpler and surely not better. This implementation contains at least two bugs, one of which is directly caused by the complexity of Coroutines. Try to find these bugs, and, maybe, others. You can find the discussion of the bug related to Coroutines in the comments section after the post (I failed to notice it myself). So, would you prefer nice-looking concurrent code, or reliable and maintainable one?

Lastly, this gist shows the implementation using Flow framework. This version of the algorithm implements “granular retry” described below, so it packs more functionality than all the preceeding implementations. In my opinion, it looks like a more complicated version of a solution with RxJava. It’s too early to say anything specific about Flow framework, but, as of today, I wouldn’t use it in production code.

More Granular Retry

During the discussion with other developers, I’ve been asked to implement one additional requirement: more granular retry mechanism. For example, if the upload fails, we can just retry the upload and there is really no need to re-process the files.

This is the result:

public class UploadFilesUseCase extends BaseBusyObservable<UploadFilesUseCase.Listener> {

    public interface Listener {
        void onFilesUploaded();
        void onFilesUploadFailed();
    }

    private final static int MAX_RETRIES = 3;

    private final Handler uiHandler = new Handler(Looper.getMainLooper());

    public void uploadFiles() {
        if (!isFreeAndBecomeBusy()) {
            // log concurrent invocation attempt
            return;
        }
        new Thread(this::uploadFilesSync).start();
    }

    @WorkerThread
    private void uploadFilesSync() {
        final AtomicReference<File> mergedA = new AtomicReference<>(null);
        final AtomicReference<File> mergedB = new AtomicReference<>(null);

        final CountDownLatch countDownLatch = new CountDownLatch(2);

        new Thread(() -> {
            mergedA.set(processAndMergeFilesOfTypeAWithRetry());
            countDownLatch.countDown();
        }).start();

        new Thread(() -> {
            mergedB.set(processAndMergeFilesOfTypeBWithRetry());
            countDownLatch.countDown();
        }).start();

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException("unexpected interrupt");
        }

        if (mergedA.get() == null || mergedB.get() == null) {
            flowFailed();
            return;
        }

        File archive;
        try {
            archive = compressMergedFiles(mergedA.get(), mergedB.get());
        } catch (OperationFailedException e) {
            flowFailed();
            return;
        }

        if (uploadFileToServerWithRetry(archive)) {
            deleteTempDir();
            notifySuccess();
        } else {
            flowFailed();
        }
    }

    @WorkerThread
    private boolean uploadFileToServerWithRetry(File archive) {
        for (int i = 0; i < MAX_RETRIES; i++) {
            int responseCode = uploadFileToServer(archive);
            if (responseCode / 100 == 2 ) {
                return true;
            }
        }
        return false;
    }

    @WorkerThread
    private int uploadFileToServer(File archive) {

        final AtomicInteger responseCode = new AtomicInteger(0);
        final CountDownLatch countDownLatch = new CountDownLatch(1);

        HttpManager.getInstance.uploadFiles(
                archive,
                new HttpRequestListener() {
                    @Override
                    public void onDone(int code, byte[] body) {
                        responseCode.set(code);
                        countDownLatch.countDown();
                    }

                    @Override
                    public void onFailure() {
                        responseCode.set(0);
                        countDownLatch.countDown();
                    }
                }
        );

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException("unexpected interrupt");
        }

        return responseCode.get();
    }

    @WorkerThread
    private void flowFailed() {
        deleteTempDir();
        notifyFailure();
    }

    @WorkerThread
    private @Nullable File processAndMergeFilesOfTypeAWithRetry() {
        for (int i = 0; i < MAX_RETRIES; i++) {
            try {
                ...
            } catch (OperationFailedException e) {
                // log the exception
            }
        }
        return null;
    }

    @WorkerThread
    private @Nullable File processAndMergeFilesOfTypeBWithRetry() {
        for (int i = 0; i < MAX_RETRIES; i++) {
            try {
                ...
            } catch (OperationFailedException e) {
                // log the exception
            }
        }
        return null;
    }

    @WorkerThread
    private File compressMergedFiles(File fileA, File fileB) throws OperationFailedException { ... }

    @WorkerThread
    private void deleteTempDir() { ... }

    private void notifySuccess() {
        uiHandler.post(() -> {
            for (Listener listener : getListeners()) {
                listener.onFilesUploaded();
            }
            becomeNotBusy();
        });
    }

    private void notifyFailure() {
        uiHandler.post(() -> {
            for (Listener listener : getListeners()) {
                listener.onFilesUploadFailed();
            }
            becomeNotBusy();
        });
    }

}

As you see, if your code already executes on a background thread synchronously, retries become just loops.

It’s not as short and sweet as RxJava or Flow operators, that’s sure. However, every developer who undertand concurrency on JVM can maintain this code in the future. In my opinion, that’s not something you can say about RxJava or Flow solution linked above.

Conclusion

In this article I demonstrated how to implement a complex concurrent flow using just the most fundamental concurrency tools. The requirements were based on a real-world scenario that solved a real problem of a real business, though I complicated them a bit to cover more ground.

The resulting implementation is robust (unless I made a mistake, of course), readable, maintainable and very flexible. You could easily change this implementation because the code is explicit and straightforward.

By the way, I’m not against frameworks. Well, not against all of them, at least. In this article I intentionally used bare Thread class to make a point, but that’s not something I’d do in real production code. Instead, I used ThreadPoster framework there. For pure Kotlin codebase I’d use Coroutines. So, there is definitely a place for frameworks in your toolbox (especially if you want to write unit tests for concurrent code). But don’t buy into the hype and don’t be that developer who uses concurrency frameworks, but doesn’t know the principles of good design and doesn’t undersand concurrency fundamentals.

Lastly, if you found the concurrency details in this post a bit challenging, check out my multithreading course. It’s the best resource to learn about concurrency for Android developers. In fact, I’m sure that all developers who work on JVM will find it useful.

Thanks for reading and don’t forget to subscribe for notifications about new posts.