Hands-On Reactive Programming with Python

Event-driven development unraveled with RxPY

Romain Picard

• Publisher: Packt Publishing
• Year: 2018
• Print ISBN: 9781789138726
• eBook ISBN: 9781789132755
• Edition: 1
• Topic: Computer Science
• Subtopic: Programming
• Index: Computer Science

A comprehensive guide to help you understand the principles of Reactive and asynchronous programming and its benefits

Key Features

• Explore the advantages of Reactive programming
• Use concurrency and parallelism in RxPY to build powerful reactive applications
• Deploy and scale your reactive applications using Docker

Book Description

Reactive programming is central to many concurrent systems, but it's famous for its steep learning curve, which makes most developers feel like they're hitting a wall. With this book, you will get to grips with reactive programming by steadily exploring various concepts

This hands-on guide gets you started with Reactive Programming (RP) in Python. You will learn abouta the principles and benefits of using RP, which can be leveraged to build powerful concurrent applications. As you progress through the chapters, you will be introduced to the paradigm of Functional and Reactive Programming (FaRP), observables and observers, and concurrency and parallelism. The book will then take you through the implementation of an audio transcoding server and introduce you to a library that helps in the writing of FaRP code. You will understand how to use third-party services and dynamically reconfigure an application.

By the end of the book, you will also have learned how to deploy and scale your applications with Docker and Traefik and explore the significant potential behind the reactive streams concept, and you'll have got to grips with a comprehensive set of best practices.

What you will learn

• Structure Python code for better readability, testing, and performance
• Explore the world of event-based programming
• Grasp the use of the most common operators in Rx
• Understand reactive extensions beyond simple examples
• Master the art of writing reusable components
• Deploy an application on a cloud platform with Docker and Traefik

Who this book is for

If you are a Python developer who wants to learn Reactive programming to build powerful concurrent and asynchronous applications, this book is for you. Basic understanding of the Python language is all you need to understand the concepts covered in this book.

Operators in RxPY

The previous chapters covered more than 30 operators. These operators allow us to implement a realistic application, but many others are often needed. This chapter describes more than 40 other operators that are also commonly used, grouped by functional category. As usual, each operator is described with its marble diagram, a comprehensive description, and an example of how to use it to help you decide when to use what operator.

The following topics will be covered in this chapter:

• Transforming observables
• Filtering observables
• Combining observables
• Utility operators
• Conditional operators
• Mathematical operators

Transforming observables

The following operators allow us to make some transformations on the items emitted on source observables. Some of them have already been covered in the previous chapters. They are the following:

Operator	Description	Chapter
map	Maps each item to another value	1
flat_map	Maps each item to an observable and flattens all emitted items in a single observable	6

The buffer operator

The buffer operator groups items emitted on the source observable based on different types of window selectors. These selectors can be defined via another observable, an item count, or time information. The following figure shows the marble diagram of the buffer operator:

Figure 9.1: The buffer operator

This operator has several prototypes:

Observable.buffer(self, buffer_openings=None, 
 buffer_closing_selector=None)
Observable.buffer_with_count(self, count, skip=None)
Observable.buffer_with_time(self, timespan, 
 timeshift=None, scheduler=None)
Observable.buffer_with_time_or_count(self, 
 timespan, count, scheduler=None)

The first version allows us to use another observable to control when a new buffer begins. Each time the buffer_openings observable emits an item, then the current buffer is emitted, and a new buffer is created. The buffer_closing_selector parameter is an optional parameter. When provided, it must be a function that returns an observable whose completion ends the current buffer. On completion, this function will be called again to start a new buffer. The buffer_closing_selector parameter can also be provided as the first parameter of this operator. If the first parameter is an observable, then it is used as the buffer_openings parameters, and if it is a function, then it is used as the buffer_closing_selector parameter.

The following is the first way to use the buffer operator:

numbers = Subject()
windows = Subject()
numbers.buffer(windows).subscribe(
 on_next = lambda i: print("on_next {}".format(i)),
 on_error = lambda e: print("on_error: {}".format(e)),
 on_completed = lambda: print("on_completed")
)

numbers.on_next(1)
numbers.on_next(2)
windows.on_next(True)
numbers.on_next(3)
numbers.on_next(4)
numbers.on_next(5)
windows.on_next(True)

The numbers observable is the observable that must be buffered. The windows observable emits items each time a new buffering window must start. The following example gives the following result:

on_next [1, 2]
on_next [3, 4, 5]

The first two items are emitted as a single list item. Then, the next three items are emitted as a second item.

The closing_selector parameter can be used in the following way:

window_selector = None
def closing_selector():
 print("closing_selector")
 global window_selector
 window_selector = Subject()
 return window_selector

numbers = Subject()
numbers.buffer(closing_selector).subscribe(
 on_next = lambda i: print("on_next {}".format(i)),
 on_error = lambda e: print("on_error: {}".format(e)),
 on_completed = lambda: print("on_completed")
)

numbers.on_next(1)
numbers.on_next(2)
numbers.on_next(3)
window_selector.on_completed()
numbers.on_next(4)
numbers.on_next(5)
window_selector.on_completed()

The numbers observable is the observable that must be buffered. The closing_selector function returns an observable that emits no items. This observable is completed after emitting three items, and then two items. This example gives the following result:

closing_selector on_next [1, 2, 3] closing_selector on_next [4, 5] closing_selector

First, the closing_selector function is called. This allows the buffer operator to subscribe to its completion. When the window_selector observable completes, then the first item is emitted (the 1, 2, 3 list). After that, ...

Table of contents

• Title Page
• Copyright and Credits
• Dedication
• Packt Upsell
• Contributors
• Preface
• An Introduction to Reactive Programming
• Asynchronous Programming in Python
• Functional Programming with ReactiveX
• Exploring Observables and Observers
• Concurrency and Parallelism in RxPY
• Implementation of an Audio Transcoding Server
• Using Third-Party Services
• Dynamic Reconfiguration and Error Management
• Operators in RxPY
• Testing and Debugging
• Deploying and Scaling Your Application
• Reactive Streams for Remote Communication
• A Checklist of Best Practices
• Assessments
• Other Books You May Enjoy