We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Firehose fails without S3DestinationConfiguration.Prefix 
Despite the AWS SDK not requiring a prefix, it seems like localstack fails without one:

```2018-05-14T17:28:43:ERROR:firehose_api: Exception on / [POST]
Traceback (most recent call last):
  File "/opt/code/localstack/.venv/lib/python2.7/site-packages/flask/app.py", line 1817, in wsgi_app
    response = self.full_dispatch_request()
  File "/opt/code/localstack/.venv/lib/python2.7/site-packages/flask/app.py", line 1477, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/opt/code/localstack/.venv/lib/python2.7/site-packages/flask_cors/extension.py", line 161, in wrapped_function
    return cors_after_request(app.make_response(f(*args, **kwargs)))
  File "/opt/code/localstack/.venv/lib/python2.7/site-packages/flask/app.py", line 1381, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/opt/code/localstack/.venv/lib/python2.7/site-packages/flask/app.py", line 1475, in full_dispatch_request
    rv = self.dispatch_request()
  File "/opt/code/localstack/.venv/lib/python2.7/site-packages/flask/app.py", line 1461, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/opt/code/localstack/localstack/services/firehose/firehose_api.py", line 183, in post_request
    put_record(stream_name, record)
  File "/opt/code/localstack/localstack/services/firehose/firehose_api.py", line 40, in put_record
    return put_records(stream_name, [record])
  File "/opt/code/localstack/localstack/services/firehose/firehose_api.py", line 63, in put_records
    prefix = s3_dest['Prefix']
KeyError: 'Prefix'```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `localstack/services/firehose/firehose_api.py`
Content:
```
1 from __future__ import print_function
2 
3 import json
4 import uuid
5 import time
6 import logging
7 import base64
8 import traceback
9 from flask import Flask, jsonify, request
10 from six import iteritems
11 from localstack.constants import TEST_AWS_ACCOUNT_ID
12 from localstack.services import generic_proxy
13 from localstack.utils.common import short_uid, to_str
14 from localstack.utils.aws import aws_responses
15 from localstack.utils.aws.aws_stack import get_s3_client, firehose_stream_arn, connect_elasticsearch
16 from boto3.dynamodb.types import TypeDeserializer
17 
18 APP_NAME = 'firehose_api'
19 app = Flask(APP_NAME)
20 ACTION_HEADER_PREFIX = 'Firehose_20150804'
21 
22 # logger
23 LOG = logging.getLogger(__name__)
24 
25 # maps stream names to details
26 DELIVERY_STREAMS = {}
27 
28 # dynamodb deserializer
29 deser = TypeDeserializer()
30 
31 
32 def get_delivery_stream_names():
33     names = []
34     for name, stream in iteritems(DELIVERY_STREAMS):
35         names.append(stream['DeliveryStreamName'])
36     return names
37 
38 
39 def put_record(stream_name, record):
40     return put_records(stream_name, [record])
41 
42 
43 def put_records(stream_name, records):
44     stream = get_stream(stream_name)
45     for dest in stream['Destinations']:
46         if 'ESDestinationDescription' in dest:
47             es_dest = dest['ESDestinationDescription']
48             es_index = es_dest['IndexName']
49             es_type = es_dest['TypeName']
50             es = connect_elasticsearch()
51             for record in records:
52                 obj_id = uuid.uuid4()
53                 data = base64.b64decode(record['Data'])
54                 body = json.loads(data)
55                 try:
56                     es.create(index=es_index, doc_type=es_type, id=obj_id, body=body)
57                 except Exception as e:
58                     LOG.error('Unable to put record to stream: %s %s' % (e, traceback.format_exc()))
59                     raise e
60         if 'S3DestinationDescription' in dest:
61             s3_dest = dest['S3DestinationDescription']
62             bucket = bucket_name(s3_dest['BucketARN'])
63             prefix = s3_dest['Prefix']
64             s3 = get_s3_client()
65             for record in records:
66                 data = base64.b64decode(record['Data'])
67                 obj_name = str(uuid.uuid4())
68                 obj_path = '%s%s' % (prefix, obj_name)
69                 try:
70                     s3.Object(bucket, obj_path).put(Body=data)
71                 except Exception as e:
72                     LOG.error('Unable to put record to stream: %s %s' % (e, traceback.format_exc()))
73                     raise e
74 
75 
76 def get_destination(stream_name, destination_id):
77     stream = get_stream(stream_name)
78     destinations = stream['Destinations']
79     for dest in destinations:
80         if dest['DestinationId'] == destination_id:
81             return dest
82     dest = {}
83     dest['DestinationId'] = destination_id
84     destinations.append(dest)
85     return dest
86 
87 
88 def update_destination(stream_name, destination_id,
89         s3_update=None, elasticsearch_update=None, version_id=None):
90     dest = get_destination(stream_name, destination_id)
91     if elasticsearch_update:
92         if 'ESDestinationDescription' not in dest:
93             dest['ESDestinationDescription'] = {}
94         for k, v in iteritems(elasticsearch_update):
95             dest['ESDestinationDescription'][k] = v
96     if s3_update:
97         if 'S3DestinationDescription' not in dest:
98             dest['S3DestinationDescription'] = {}
99         for k, v in iteritems(s3_update):
100             dest['S3DestinationDescription'][k] = v
101     return dest
102 
103 
104 def create_stream(stream_name, s3_destination=None, elasticsearch_destination=None):
105     stream = {
106         'HasMoreDestinations': False,
107         'VersionId': '1',
108         'CreateTimestamp': time.time(),
109         'DeliveryStreamARN': firehose_stream_arn(stream_name),
110         'DeliveryStreamStatus': 'ACTIVE',
111         'DeliveryStreamName': stream_name,
112         'Destinations': []
113     }
114     DELIVERY_STREAMS[stream_name] = stream
115     if elasticsearch_destination:
116         update_destination(stream_name=stream_name,
117             destination_id=short_uid(),
118             elasticsearch_update=elasticsearch_destination)
119     if s3_destination:
120         update_destination(stream_name=stream_name, destination_id=short_uid(), s3_update=s3_destination)
121     return stream
122 
123 
124 def delete_stream(stream_name):
125     stream = DELIVERY_STREAMS.pop(stream_name, {})
126     if not stream:
127         return error_not_found(stream_name)
128     return {}
129 
130 
131 def get_stream(stream_name):
132     if stream_name not in DELIVERY_STREAMS:
133         return None
134     return DELIVERY_STREAMS[stream_name]
135 
136 
137 def bucket_name(bucket_arn):
138     return bucket_arn.split(':::')[-1]
139 
140 
141 def role_arn(stream_name):
142     return 'arn:aws:iam::%s:role/%s' % (TEST_AWS_ACCOUNT_ID, stream_name)
143 
144 
145 def error_not_found(stream_name):
146     msg = 'Firehose %s under account %s not found.' % (stream_name, TEST_AWS_ACCOUNT_ID)
147     return error_response(msg, code=400, error_type='ResourceNotFoundException')
148 
149 
150 def error_response(msg, code=500, error_type='InternalFailure'):
151     return aws_responses.flask_error_response(msg, code=code, error_type=error_type)
152 
153 
154 @app.route('/', methods=['POST'])
155 def post_request():
156     action = request.headers.get('x-amz-target')
157     data = json.loads(to_str(request.data))
158     response = None
159     if action == '%s.ListDeliveryStreams' % ACTION_HEADER_PREFIX:
160         response = {
161             'DeliveryStreamNames': get_delivery_stream_names(),
162             'HasMoreDeliveryStreams': False
163         }
164     elif action == '%s.CreateDeliveryStream' % ACTION_HEADER_PREFIX:
165         stream_name = data['DeliveryStreamName']
166         response = create_stream(stream_name,
167             s3_destination=data.get('S3DestinationConfiguration'),
168             elasticsearch_destination=data.get('ElasticsearchDestinationConfiguration'))
169     elif action == '%s.DeleteDeliveryStream' % ACTION_HEADER_PREFIX:
170         stream_name = data['DeliveryStreamName']
171         response = delete_stream(stream_name)
172     elif action == '%s.DescribeDeliveryStream' % ACTION_HEADER_PREFIX:
173         stream_name = data['DeliveryStreamName']
174         response = get_stream(stream_name)
175         if not response:
176             return error_not_found(stream_name)
177         response = {
178             'DeliveryStreamDescription': response
179         }
180     elif action == '%s.PutRecord' % ACTION_HEADER_PREFIX:
181         stream_name = data['DeliveryStreamName']
182         record = data['Record']
183         put_record(stream_name, record)
184         response = {
185             'RecordId': str(uuid.uuid4())
186         }
187     elif action == '%s.PutRecordBatch' % ACTION_HEADER_PREFIX:
188         stream_name = data['DeliveryStreamName']
189         records = data['Records']
190         put_records(stream_name, records)
191         response = {
192             'FailedPutCount': 0,
193             'RequestResponses': []
194         }
195     elif action == '%s.UpdateDestination' % ACTION_HEADER_PREFIX:
196         stream_name = data['DeliveryStreamName']
197         version_id = data['CurrentDeliveryStreamVersionId']
198         destination_id = data['DestinationId']
199         s3_update = data['S3DestinationUpdate'] if 'S3DestinationUpdate' in data else None
200         update_destination(stream_name=stream_name, destination_id=destination_id,
201             s3_update=s3_update, version_id=version_id)
202         es_update = data['ESDestinationUpdate'] if 'ESDestinationUpdate' in data else None
203         update_destination(stream_name=stream_name, destination_id=destination_id,
204             es_update=es_update, version_id=version_id)
205         response = {}
206     else:
207         response = error_response('Unknown action "%s"' % action, code=400, error_type='InvalidAction')
208 
209     if isinstance(response, dict):
210         response = jsonify(response)
211     return response
212 
213 
214 def serve(port, quiet=True):
215     generic_proxy.serve_flask_app(app=app, port=port, quiet=quiet)
216
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/localstack/services/firehose/firehose_api.py b/localstack/services/firehose/firehose_api.py
--- a/localstack/services/firehose/firehose_api.py
+++ b/localstack/services/firehose/firehose_api.py
@@ -60,12 +60,12 @@
         if 'S3DestinationDescription' in dest:
             s3_dest = dest['S3DestinationDescription']
             bucket = bucket_name(s3_dest['BucketARN'])
-            prefix = s3_dest['Prefix']
+            prefix = s3_dest.get('Prefix', '')
             s3 = get_s3_client()
             for record in records:
                 data = base64.b64decode(record['Data'])
                 obj_name = str(uuid.uuid4())
-                obj_path = '%s%s' % (prefix, obj_name)
+                obj_path = '%s%s%s' % (prefix, '' if prefix.endswith('/') else '/', obj_name)
                 try:
                     s3.Object(bucket, obj_path).put(Body=data)
                 except Exception as e:

{"golden_diff": "diff --git a/localstack/services/firehose/firehose_api.py b/localstack/services/firehose/firehose_api.py\n--- a/localstack/services/firehose/firehose_api.py\n+++ b/localstack/services/firehose/firehose_api.py\n@@ -60,12 +60,12 @@\n         if 'S3DestinationDescription' in dest:\n             s3_dest = dest['S3DestinationDescription']\n             bucket = bucket_name(s3_dest['BucketARN'])\n-            prefix = s3_dest['Prefix']\n+            prefix = s3_dest.get('Prefix', '')\n             s3 = get_s3_client()\n             for record in records:\n                 data = base64.b64decode(record['Data'])\n                 obj_name = str(uuid.uuid4())\n-                obj_path = '%s%s' % (prefix, obj_name)\n+                obj_path = '%s%s%s' % (prefix, '' if prefix.endswith('/') else '/', obj_name)\n                 try:\n                     s3.Object(bucket, obj_path).put(Body=data)\n                 except Exception as e:\n", "issue": "Firehose fails without S3DestinationConfiguration.Prefix \nDespite the AWS SDK not requiring a prefix, it seems like localstack fails without one:\r\n\r\n```2018-05-14T17:28:43:ERROR:firehose_api: Exception on / [POST]\r\nTraceback (most recent call last):\r\n  File \"/opt/code/localstack/.venv/lib/python2.7/site-packages/flask/app.py\", line 1817, in wsgi_app\r\n    response = self.full_dispatch_request()\r\n  File \"/opt/code/localstack/.venv/lib/python2.7/site-packages/flask/app.py\", line 1477, in full_dispatch_request\r\n    rv = self.handle_user_exception(e)\r\n  File \"/opt/code/localstack/.venv/lib/python2.7/site-packages/flask_cors/extension.py\", line 161, in wrapped_function\r\n    return cors_after_request(app.make_response(f(*args, **kwargs)))\r\n  File \"/opt/code/localstack/.venv/lib/python2.7/site-packages/flask/app.py\", line 1381, in handle_user_exception\r\n    reraise(exc_type, exc_value, tb)\r\n  File \"/opt/code/localstack/.venv/lib/python2.7/site-packages/flask/app.py\", line 1475, in full_dispatch_request\r\n    rv = self.dispatch_request()\r\n  File \"/opt/code/localstack/.venv/lib/python2.7/site-packages/flask/app.py\", line 1461, in dispatch_request\r\n    return self.view_functions[rule.endpoint](**req.view_args)\r\n  File \"/opt/code/localstack/localstack/services/firehose/firehose_api.py\", line 183, in post_request\r\n    put_record(stream_name, record)\r\n  File \"/opt/code/localstack/localstack/services/firehose/firehose_api.py\", line 40, in put_record\r\n    return put_records(stream_name, [record])\r\n  File \"/opt/code/localstack/localstack/services/firehose/firehose_api.py\", line 63, in put_records\r\n    prefix = s3_dest['Prefix']\r\nKeyError: 'Prefix'```\n", "before_files": [{"content": "from __future__ import print_function\n\nimport json\nimport uuid\nimport time\nimport logging\nimport base64\nimport traceback\nfrom flask import Flask, jsonify, request\nfrom six import iteritems\nfrom localstack.constants import TEST_AWS_ACCOUNT_ID\nfrom localstack.services import generic_proxy\nfrom localstack.utils.common import short_uid, to_str\nfrom localstack.utils.aws import aws_responses\nfrom localstack.utils.aws.aws_stack import get_s3_client, firehose_stream_arn, connect_elasticsearch\nfrom boto3.dynamodb.types import TypeDeserializer\n\nAPP_NAME = 'firehose_api'\napp = Flask(APP_NAME)\nACTION_HEADER_PREFIX = 'Firehose_20150804'\n\n# logger\nLOG = logging.getLogger(__name__)\n\n# maps stream names to details\nDELIVERY_STREAMS = {}\n\n# dynamodb deserializer\ndeser = TypeDeserializer()\n\n\ndef get_delivery_stream_names():\n    names = []\n    for name, stream in iteritems(DELIVERY_STREAMS):\n        names.append(stream['DeliveryStreamName'])\n    return names\n\n\ndef put_record(stream_name, record):\n    return put_records(stream_name, [record])\n\n\ndef put_records(stream_name, records):\n    stream = get_stream(stream_name)\n    for dest in stream['Destinations']:\n        if 'ESDestinationDescription' in dest:\n            es_dest = dest['ESDestinationDescription']\n            es_index = es_dest['IndexName']\n            es_type = es_dest['TypeName']\n            es = connect_elasticsearch()\n            for record in records:\n                obj_id = uuid.uuid4()\n                data = base64.b64decode(record['Data'])\n                body = json.loads(data)\n                try:\n                    es.create(index=es_index, doc_type=es_type, id=obj_id, body=body)\n                except Exception as e:\n                    LOG.error('Unable to put record to stream: %s %s' % (e, traceback.format_exc()))\n                    raise e\n        if 'S3DestinationDescription' in dest:\n            s3_dest = dest['S3DestinationDescription']\n            bucket = bucket_name(s3_dest['BucketARN'])\n            prefix = s3_dest['Prefix']\n            s3 = get_s3_client()\n            for record in records:\n                data = base64.b64decode(record['Data'])\n                obj_name = str(uuid.uuid4())\n                obj_path = '%s%s' % (prefix, obj_name)\n                try:\n                    s3.Object(bucket, obj_path).put(Body=data)\n                except Exception as e:\n                    LOG.error('Unable to put record to stream: %s %s' % (e, traceback.format_exc()))\n                    raise e\n\n\ndef get_destination(stream_name, destination_id):\n    stream = get_stream(stream_name)\n    destinations = stream['Destinations']\n    for dest in destinations:\n        if dest['DestinationId'] == destination_id:\n            return dest\n    dest = {}\n    dest['DestinationId'] = destination_id\n    destinations.append(dest)\n    return dest\n\n\ndef update_destination(stream_name, destination_id,\n        s3_update=None, elasticsearch_update=None, version_id=None):\n    dest = get_destination(stream_name, destination_id)\n    if elasticsearch_update:\n        if 'ESDestinationDescription' not in dest:\n            dest['ESDestinationDescription'] = {}\n        for k, v in iteritems(elasticsearch_update):\n            dest['ESDestinationDescription'][k] = v\n    if s3_update:\n        if 'S3DestinationDescription' not in dest:\n            dest['S3DestinationDescription'] = {}\n        for k, v in iteritems(s3_update):\n            dest['S3DestinationDescription'][k] = v\n    return dest\n\n\ndef create_stream(stream_name, s3_destination=None, elasticsearch_destination=None):\n    stream = {\n        'HasMoreDestinations': False,\n        'VersionId': '1',\n        'CreateTimestamp': time.time(),\n        'DeliveryStreamARN': firehose_stream_arn(stream_name),\n        'DeliveryStreamStatus': 'ACTIVE',\n        'DeliveryStreamName': stream_name,\n        'Destinations': []\n    }\n    DELIVERY_STREAMS[stream_name] = stream\n    if elasticsearch_destination:\n        update_destination(stream_name=stream_name,\n            destination_id=short_uid(),\n            elasticsearch_update=elasticsearch_destination)\n    if s3_destination:\n        update_destination(stream_name=stream_name, destination_id=short_uid(), s3_update=s3_destination)\n    return stream\n\n\ndef delete_stream(stream_name):\n    stream = DELIVERY_STREAMS.pop(stream_name, {})\n    if not stream:\n        return error_not_found(stream_name)\n    return {}\n\n\ndef get_stream(stream_name):\n    if stream_name not in DELIVERY_STREAMS:\n        return None\n    return DELIVERY_STREAMS[stream_name]\n\n\ndef bucket_name(bucket_arn):\n    return bucket_arn.split(':::')[-1]\n\n\ndef role_arn(stream_name):\n    return 'arn:aws:iam::%s:role/%s' % (TEST_AWS_ACCOUNT_ID, stream_name)\n\n\ndef error_not_found(stream_name):\n    msg = 'Firehose %s under account %s not found.' % (stream_name, TEST_AWS_ACCOUNT_ID)\n    return error_response(msg, code=400, error_type='ResourceNotFoundException')\n\n\ndef error_response(msg, code=500, error_type='InternalFailure'):\n    return aws_responses.flask_error_response(msg, code=code, error_type=error_type)\n\n\n@app.route('/', methods=['POST'])\ndef post_request():\n    action = request.headers.get('x-amz-target')\n    data = json.loads(to_str(request.data))\n    response = None\n    if action == '%s.ListDeliveryStreams' % ACTION_HEADER_PREFIX:\n        response = {\n            'DeliveryStreamNames': get_delivery_stream_names(),\n            'HasMoreDeliveryStreams': False\n        }\n    elif action == '%s.CreateDeliveryStream' % ACTION_HEADER_PREFIX:\n        stream_name = data['DeliveryStreamName']\n        response = create_stream(stream_name,\n            s3_destination=data.get('S3DestinationConfiguration'),\n            elasticsearch_destination=data.get('ElasticsearchDestinationConfiguration'))\n    elif action == '%s.DeleteDeliveryStream' % ACTION_HEADER_PREFIX:\n        stream_name = data['DeliveryStreamName']\n        response = delete_stream(stream_name)\n    elif action == '%s.DescribeDeliveryStream' % ACTION_HEADER_PREFIX:\n        stream_name = data['DeliveryStreamName']\n        response = get_stream(stream_name)\n        if not response:\n            return error_not_found(stream_name)\n        response = {\n            'DeliveryStreamDescription': response\n        }\n    elif action == '%s.PutRecord' % ACTION_HEADER_PREFIX:\n        stream_name = data['DeliveryStreamName']\n        record = data['Record']\n        put_record(stream_name, record)\n        response = {\n            'RecordId': str(uuid.uuid4())\n        }\n    elif action == '%s.PutRecordBatch' % ACTION_HEADER_PREFIX:\n        stream_name = data['DeliveryStreamName']\n        records = data['Records']\n        put_records(stream_name, records)\n        response = {\n            'FailedPutCount': 0,\n            'RequestResponses': []\n        }\n    elif action == '%s.UpdateDestination' % ACTION_HEADER_PREFIX:\n        stream_name = data['DeliveryStreamName']\n        version_id = data['CurrentDeliveryStreamVersionId']\n        destination_id = data['DestinationId']\n        s3_update = data['S3DestinationUpdate'] if 'S3DestinationUpdate' in data else None\n        update_destination(stream_name=stream_name, destination_id=destination_id,\n            s3_update=s3_update, version_id=version_id)\n        es_update = data['ESDestinationUpdate'] if 'ESDestinationUpdate' in data else None\n        update_destination(stream_name=stream_name, destination_id=destination_id,\n            es_update=es_update, version_id=version_id)\n        response = {}\n    else:\n        response = error_response('Unknown action \"%s\"' % action, code=400, error_type='InvalidAction')\n\n    if isinstance(response, dict):\n        response = jsonify(response)\n    return response\n\n\ndef serve(port, quiet=True):\n    generic_proxy.serve_flask_app(app=app, port=port, quiet=quiet)\n", "path": "localstack/services/firehose/firehose_api.py"}], "after_files": [{"content": "from __future__ import print_function\n\nimport json\nimport uuid\nimport time\nimport logging\nimport base64\nimport traceback\nfrom flask import Flask, jsonify, request\nfrom six import iteritems\nfrom localstack.constants import TEST_AWS_ACCOUNT_ID\nfrom localstack.services import generic_proxy\nfrom localstack.utils.common import short_uid, to_str\nfrom localstack.utils.aws import aws_responses\nfrom localstack.utils.aws.aws_stack import get_s3_client, firehose_stream_arn, connect_elasticsearch\nfrom boto3.dynamodb.types import TypeDeserializer\n\nAPP_NAME = 'firehose_api'\napp = Flask(APP_NAME)\nACTION_HEADER_PREFIX = 'Firehose_20150804'\n\n# logger\nLOG = logging.getLogger(__name__)\n\n# maps stream names to details\nDELIVERY_STREAMS = {}\n\n# dynamodb deserializer\ndeser = TypeDeserializer()\n\n\ndef get_delivery_stream_names():\n    names = []\n    for name, stream in iteritems(DELIVERY_STREAMS):\n        names.append(stream['DeliveryStreamName'])\n    return names\n\n\ndef put_record(stream_name, record):\n    return put_records(stream_name, [record])\n\n\ndef put_records(stream_name, records):\n    stream = get_stream(stream_name)\n    for dest in stream['Destinations']:\n        if 'ESDestinationDescription' in dest:\n            es_dest = dest['ESDestinationDescription']\n            es_index = es_dest['IndexName']\n            es_type = es_dest['TypeName']\n            es = connect_elasticsearch()\n            for record in records:\n                obj_id = uuid.uuid4()\n                data = base64.b64decode(record['Data'])\n                body = json.loads(data)\n                try:\n                    es.create(index=es_index, doc_type=es_type, id=obj_id, body=body)\n                except Exception as e:\n                    LOG.error('Unable to put record to stream: %s %s' % (e, traceback.format_exc()))\n                    raise e\n        if 'S3DestinationDescription' in dest:\n            s3_dest = dest['S3DestinationDescription']\n            bucket = bucket_name(s3_dest['BucketARN'])\n            prefix = s3_dest.get('Prefix', '')\n            s3 = get_s3_client()\n            for record in records:\n                data = base64.b64decode(record['Data'])\n                obj_name = str(uuid.uuid4())\n                obj_path = '%s%s%s' % (prefix, '' if prefix.endswith('/') else '/', obj_name)\n                try:\n                    s3.Object(bucket, obj_path).put(Body=data)\n                except Exception as e:\n                    LOG.error('Unable to put record to stream: %s %s' % (e, traceback.format_exc()))\n                    raise e\n\n\ndef get_destination(stream_name, destination_id):\n    stream = get_stream(stream_name)\n    destinations = stream['Destinations']\n    for dest in destinations:\n        if dest['DestinationId'] == destination_id:\n            return dest\n    dest = {}\n    dest['DestinationId'] = destination_id\n    destinations.append(dest)\n    return dest\n\n\ndef update_destination(stream_name, destination_id,\n        s3_update=None, elasticsearch_update=None, version_id=None):\n    dest = get_destination(stream_name, destination_id)\n    if elasticsearch_update:\n        if 'ESDestinationDescription' not in dest:\n            dest['ESDestinationDescription'] = {}\n        for k, v in iteritems(elasticsearch_update):\n            dest['ESDestinationDescription'][k] = v\n    if s3_update:\n        if 'S3DestinationDescription' not in dest:\n            dest['S3DestinationDescription'] = {}\n        for k, v in iteritems(s3_update):\n            dest['S3DestinationDescription'][k] = v\n    return dest\n\n\ndef create_stream(stream_name, s3_destination=None, elasticsearch_destination=None):\n    stream = {\n        'HasMoreDestinations': False,\n        'VersionId': '1',\n        'CreateTimestamp': time.time(),\n        'DeliveryStreamARN': firehose_stream_arn(stream_name),\n        'DeliveryStreamStatus': 'ACTIVE',\n        'DeliveryStreamName': stream_name,\n        'Destinations': []\n    }\n    DELIVERY_STREAMS[stream_name] = stream\n    if elasticsearch_destination:\n        update_destination(stream_name=stream_name,\n            destination_id=short_uid(),\n            elasticsearch_update=elasticsearch_destination)\n    if s3_destination:\n        update_destination(stream_name=stream_name, destination_id=short_uid(), s3_update=s3_destination)\n    return stream\n\n\ndef delete_stream(stream_name):\n    stream = DELIVERY_STREAMS.pop(stream_name, {})\n    if not stream:\n        return error_not_found(stream_name)\n    return {}\n\n\ndef get_stream(stream_name):\n    if stream_name not in DELIVERY_STREAMS:\n        return None\n    return DELIVERY_STREAMS[stream_name]\n\n\ndef bucket_name(bucket_arn):\n    return bucket_arn.split(':::')[-1]\n\n\ndef role_arn(stream_name):\n    return 'arn:aws:iam::%s:role/%s' % (TEST_AWS_ACCOUNT_ID, stream_name)\n\n\ndef error_not_found(stream_name):\n    msg = 'Firehose %s under account %s not found.' % (stream_name, TEST_AWS_ACCOUNT_ID)\n    return error_response(msg, code=400, error_type='ResourceNotFoundException')\n\n\ndef error_response(msg, code=500, error_type='InternalFailure'):\n    return aws_responses.flask_error_response(msg, code=code, error_type=error_type)\n\n\n@app.route('/', methods=['POST'])\ndef post_request():\n    action = request.headers.get('x-amz-target')\n    data = json.loads(to_str(request.data))\n    response = None\n    if action == '%s.ListDeliveryStreams' % ACTION_HEADER_PREFIX:\n        response = {\n            'DeliveryStreamNames': get_delivery_stream_names(),\n            'HasMoreDeliveryStreams': False\n        }\n    elif action == '%s.CreateDeliveryStream' % ACTION_HEADER_PREFIX:\n        stream_name = data['DeliveryStreamName']\n        response = create_stream(stream_name,\n            s3_destination=data.get('S3DestinationConfiguration'),\n            elasticsearch_destination=data.get('ElasticsearchDestinationConfiguration'))\n    elif action == '%s.DeleteDeliveryStream' % ACTION_HEADER_PREFIX:\n        stream_name = data['DeliveryStreamName']\n        response = delete_stream(stream_name)\n    elif action == '%s.DescribeDeliveryStream' % ACTION_HEADER_PREFIX:\n        stream_name = data['DeliveryStreamName']\n        response = get_stream(stream_name)\n        if not response:\n            return error_not_found(stream_name)\n        response = {\n            'DeliveryStreamDescription': response\n        }\n    elif action == '%s.PutRecord' % ACTION_HEADER_PREFIX:\n        stream_name = data['DeliveryStreamName']\n        record = data['Record']\n        put_record(stream_name, record)\n        response = {\n            'RecordId': str(uuid.uuid4())\n        }\n    elif action == '%s.PutRecordBatch' % ACTION_HEADER_PREFIX:\n        stream_name = data['DeliveryStreamName']\n        records = data['Records']\n        put_records(stream_name, records)\n        response = {\n            'FailedPutCount': 0,\n            'RequestResponses': []\n        }\n    elif action == '%s.UpdateDestination' % ACTION_HEADER_PREFIX:\n        stream_name = data['DeliveryStreamName']\n        version_id = data['CurrentDeliveryStreamVersionId']\n        destination_id = data['DestinationId']\n        s3_update = data['S3DestinationUpdate'] if 'S3DestinationUpdate' in data else None\n        update_destination(stream_name=stream_name, destination_id=destination_id,\n            s3_update=s3_update, version_id=version_id)\n        es_update = data['ESDestinationUpdate'] if 'ESDestinationUpdate' in data else None\n        update_destination(stream_name=stream_name, destination_id=destination_id,\n            es_update=es_update, version_id=version_id)\n        response = {}\n    else:\n        response = error_response('Unknown action \"%s\"' % action, code=400, error_type='InvalidAction')\n\n    if isinstance(response, dict):\n        response = jsonify(response)\n    return response\n\n\ndef serve(port, quiet=True):\n    generic_proxy.serve_flask_app(app=app, port=port, quiet=quiet)\n", "path": "localstack/services/firehose/firehose_api.py"}]}