The final step is to use the inotify driver in the application. It must first be added in the driver, source, and sink of the application entry point:

Drivers = namedtuple('Drivers', 
    ['encoder', 'httpd', 's3', 'file', 'inotify', 'argv'])
Source = namedtuple('Source', 
    ['encoder', 'httpd', 's3', 'file', 'inotify', 'argv'])
Sink = namedtuple('Sink', 
    ['encoder', 'httpd', 's3', 'file', 'inotify'])

The inotify driver must be created in the constructor of the Drivers object when the run function is called. Then it can be configured:

    monitor_init = (
        parsed_argv
        .flat_map(lambda i : Observable.from_([
            inotify.AddWatch(id='config', path=i.value, 
            flags=aionotify.Flags.MODIFY),
            inotify.Start(),
        ]))
    )

An observable containing configuration and startup items is created from the parsed_argv observable. Since the parsed_argv observable contains only one item, the monitor_init observable will emit two items. This new observable is the sink of the inotify driver. The items coming from the inotify driver are then used to re-read the configuration file:

    config_update = (
        sources.inotify.response
        .debounce(5000)
        .map(lambda i: True)
        .start_with(True)
    )

    read_config_file = (
        Observable.combine_latest(parsed_argv, config_update,
            lambda config, _: file.Read(id='config', 
            path=config.value))
    )

The config_update observable emits a Boolean item each time the configuration file is changed. The debounce operator avoids events flooding. In this case, an event is emitted only when no change occurs within 5 seconds. Finally, the config_update observable starts by emitting an item immediately to trigger the initial read of the configuration file.

The read_config_file observable is no longer built directly from the argv source. Instead, it is a composition of the argv value (that is, the path of the configuration file) and the inotify events. The usage of the combine_latest operator allows you to emit a Read item with the path of the configuration file each time.

With all these changes, the configuration file is read and parsed each time it is modified. Moreover, the encode driver take into account these changes. However, there is now an issue—the config observable emits an item each time the configuration file is changed instead of only one single item. This means that the configuration of the HTTP and S3 drivers is done each time the configuration file is modified. Since these drivers do not support reconfiguration, it raises errors. This can be fixed by using only the first item emitted by the config observable when initializing these drivers:

http_init = (
    config.take(1)
    ...

s3_init = (
    config.take(1)
    ...

This is done trivially with the take operator that emits only the first n items of an observable. Here, only the first item is used. Note how this issue is the opposite of what happens with other programming paradigms. In a reactive application, handling change is the default behavior, while doing things only once requires special care. In a typical object-oriented application, adding a dynamic configuration afterwards almost always requires rewriting a significant part of the application. In a reactive application, this is a much more simple evolution.

A last sanity check can be added to avoid reconfiguring the encode driver when it is not needed. If for any reason the configuration file changed but not the encode part of it, then it is useless to reconfigure the encoder driver. This can be done with the following modifications in the encode_init observable chain:

    encode_init = (
        config
        .map(lambda i: i.encode)
        .distinct_until_changed()
        .map(lambda i: encoder.Configure(
            samplerate=i.samplerate,
            bitdepth=i.bitdepth))
    )

First, the whole configuration is mapped to just contain the encode part. Then these items are filtered to be emitted only when they change, thanks to the distinct_until_changed operator.

The new code of the transcode server is available in the GitHub repository (https://github.com/PacktPublishing/Hands-On-Reactive-Programming-with-Python) of this book, in the audio-encode-server-4 sub-folder. It creates logs when the encode driver configuration changes. So you can test it and see that the configuration is updated 5 seconds after the the configuration file is modified.