These evolutions can now be used in the server itself. This implies some evolutions in the drivers used by the audio_encoder component, as demonstrated in the following example:

import audio_encode_server.s3 as s3

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

The S3 driver has been added, and it is used both in Source and Sink of the component. Inside the component, a new step has been added between the encoding result and the sending of the HTTP response: the file is uploaded on the S3 database. This is done in the following way:

    store_requests = (
        sources.encoder.response
        .map(lambda i: s3.UploadObject(
            key=i.key + '.flac',
            data=i.data,
            id=i.id,
        ))
    )

Instead of directly being used to send the HTTP response, the response from the encoder is mapped to an S3 upload request. The result of the upload request is used to send the HTTP answer, as can be seen in the following example:

    http_response = (
        sources.s3.response
        .map(lambda i: httpd.Response(
            data='ok'.encode('utf-8'),
            context=i.id,
        ))
    )

The S3 driver must also be initialized from information stored in the configuration file. A new section is added in the configuration file to store the S3 database parameters. They typically look like the following example:

  "s3": {
    "access_key": "P0B76GIFPB6T0OD5I67U",
    "secret_key": "3+kGshnhQ7i41UabvMz8buqUKgtGsikmdL1Q+oDR",
    "bucket": "audio",
    "endpoint_url": "http://localhost:9000",
    "region_name": "eu-west-1"
  }

Thanks to the generic configuration parser that was implemented, this new section can be used directly to initialize the driver, as can be seen in the following example:

    s3_init = (
        config
        .map(lambda i: s3.Configure(
            access_key=i.s3.access_key,
            secret_key=i.s3.secret_key,
            bucket=i.s3.bucket,
            endpoint_url=i.s3.endpoint_url,
            region_name=i.s3.region_name,
        ))
    )

The last step is to merge the S3 observables and return them in the Source object. This is demonstrated in the following code:

    s3_requests = Observable.merge(s3_init, store_requests)

    return Sink(
        encoder=encoder.Sink(request=encoder_request),
        s3=s3.Sink(request=s3_requests),
        file=file.Sink(request=file_requests),
        httpd=httpd.Sink(control=http),
    )

The whole code is available in the audio-encode-server-2 file in this chapter's repository (https://github.com/PacktPublishing/Hands-On-Reactive-Programming-with-Python). This new implementation can be used in the same way as the first one, as detailed here:

(venv-rx)$ python3 setup.py install

This installs the application package, as well as all its dependencies. The server can then be started in the following way:

(venv-rx)$ audio-encode-server --config ./config.sample.json

And Curl can be used to transcode a file and store it on the S3 database, as shown in this example:

curl -X POST --data-binary @../audio-dataset/banjo1.mp3 http://localhost:8080/api/transcode/v1/flac/banjo1

After this call, a new banjo1.flac file should exist in the S3 audio bucket. You can download it via the web interface of Minio or AWS and check that it corresponds to the original MP3 file.