diff --git a/README.md b/README.md
index 6e99821..ddb9731 100644
--- a/README.md
+++ b/README.md
@@ -7,7 +7,7 @@
 [![GitHub release](https://img.shields.io/github/v/release/swisspost/vertx-rest-storage)](https://github.com/swisspost/vertx-rest-storage/releases/latest)
 [![Maven Central](https://img.shields.io/maven-central/v/org.swisspush/rest-storage.svg)]()
 
-Persistence for REST resources in the filesystem or a redis database. 
+Persistence for REST resources in the filesystem, Aws S3 storage or a redis database. 
 
 Stores resources in a hierarchical way according to their URI. It actually implements a generic CRUD REST service.
 
@@ -223,6 +223,15 @@ The following configuration values are available:
 | rejectStorageWriteOnLowMemory           | redis  | false                    | When set to _true_, PUT requests with the x-importance-level header can be rejected when memory gets low                              |
 | freeMemoryCheckIntervalMs               | redis  | 60000                    | The interval in milliseconds to calculate the actual memory usage                                                                     |
 | redisReadyCheckIntervalMs               | redis  | -1                       | The interval in milliseconds to calculate the "ready state" of redis. When value < 1, no "ready state" will be calculated             |
+| awsS3Region                             | s3     |                          | The region of AWS S3 server, with local service such localstack, also need set a valid region                                         |
+| s3BucketName                            | s3     |                          | The S3 bucket name                                                                                                                    |
+| s3AccessKeyId                           | s3     |                          | The s3 access key Id                                                                                                                  |
+| s3SecretAccessKey                       | s3     |                          | The s3 secret access key                                                                                                              |
+| localS3                                 | s3     |                          | Set to true in order to use a local S3 instance instead of AWS                                                                        |
+| localS3Endpoint                         | s3     |                          | The endpoint/host to use in case that localS3 is set to true, e.g. 127.0.0.1 (in my case it had to be an IP)                          |
+| localS3Port                             | s3     |                          | The port to use in case that localS3 is set to true, e.g. 4566                                                                        |
+| createBucketIfNotExist                  | s3     |                          | create bucket if bucket not exist, related permission required                                                                        |
+
 
 ### Configuration util
 
@@ -249,11 +258,21 @@ JsonObject json  = new ModuleConfiguration().asJsonObject();
 ```
 
 ## Storage types
-Currently, there are two storage types supported. File system storage and redis storage.
+Currently, there are three storage types supported. File system storage, S3 storage and redis storage.
 
 ### File System Storage
 The data is stored hierarchically on the file system. This is the default storage type when not overridden in the configuration.
 
+### S3 storage
+The data is stored in a S3 instance.
+
+#### AWS S3
+See https://aws.amazon.com/s3 it is also possible to use a local instance using https://docs.localstack.cloud/user-guide/aws/s3/
+
+
+docker run --rm -p 4566:4566 -v ./s3:/var/lib/localstack localstack/localstack:s3-latest
+
+
 ### Redis Storage
 The data is stored in a redis database.
 Caution: The redis storage implementation does not currently support streaming. Avoid transferring too big payloads since they will be entirely copied in memory.
diff --git a/pom.xml b/pom.xml
index 7b543d4..9c3a451 100644
--- a/pom.xml
+++ b/pom.xml
@@ -69,7 +69,11 @@
             <artifactId>guava</artifactId>
             <version>${guava.version}</version>
         </dependency>
-
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>s3</artifactId>
+            <version>${awssdk.version}</version>
+        </dependency>
         <!-- TEST dependencies -->
         <dependency>
             <!--
@@ -133,6 +137,11 @@
             <artifactId>jackson-databind</artifactId>
             <version>${jackson.version}</version>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.nio.s3</groupId>
+            <artifactId>aws-java-nio-spi-for-s3</artifactId>
+            <version>${nio.spi.s3.version}</version>
+        </dependency>
     </dependencies>
     <build>
         <pluginManagement>
@@ -404,6 +413,8 @@
     </profiles>
     <properties>
         <vertx.version>4.5.2</vertx.version>
+        <awssdk.version>2.27.17</awssdk.version>
+        <nio.spi.s3.version>2.1.0</nio.spi.s3.version>
         <slf4j.version>2.0.10</slf4j.version>
         <junit.version>4.13.2</junit.version>
         <jackson.version>2.16.0</jackson.version>
diff --git a/src/main/java/org/swisspush/reststorage/RestStorageMod.java b/src/main/java/org/swisspush/reststorage/RestStorageMod.java
index fccfb73..6347095 100644
--- a/src/main/java/org/swisspush/reststorage/RestStorageMod.java
+++ b/src/main/java/org/swisspush/reststorage/RestStorageMod.java
@@ -1,6 +1,10 @@
 package org.swisspush.reststorage;
-
-import io.vertx.core.*;
+;
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
+import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.core.http.HttpServerRequest;
 import org.slf4j.Logger;
@@ -9,6 +13,7 @@
 import org.swisspush.reststorage.redis.DefaultRedisProvider;
 import org.swisspush.reststorage.redis.RedisProvider;
 import org.swisspush.reststorage.redis.RedisStorage;
+import org.swisspush.reststorage.s3.S3FileSystemStorage;
 import org.swisspush.reststorage.util.ModuleConfiguration;
 
 import static org.swisspush.reststorage.exception.RestStorageExceptionFactory.newRestStorageThriftyExceptionFactory;
@@ -73,6 +78,14 @@ private Future<Storage> createStorage(ModuleConfiguration moduleConfiguration) {
             case filesystem:
                 promise.complete(new FileSystemStorage(vertx, exceptionFactory, moduleConfiguration.getRoot()));
                 break;
+            case s3:
+                promise.complete(new S3FileSystemStorage(vertx, exceptionFactory, moduleConfiguration.getRoot(),
+                        moduleConfiguration.getAwsS3Region(), moduleConfiguration.getS3BucketName(),
+                        moduleConfiguration.getS3AccessKeyId(), moduleConfiguration.getS3SecretAccessKey(),
+                        moduleConfiguration.getS3UseTlsConnection(), moduleConfiguration.isLocalS3(),
+                        moduleConfiguration.getLocalS3Endpoint(), moduleConfiguration.getLocalS3Port(),
+                        moduleConfiguration.getCreateBucketIfNotPresentYet()));
+                break;
             case redis:
                 createRedisStorage(vertx, moduleConfiguration).onComplete(event -> {
                     if(event.succeeded()){
diff --git a/src/main/java/org/swisspush/reststorage/s3/FileReadStream.java b/src/main/java/org/swisspush/reststorage/s3/FileReadStream.java
new file mode 100644
index 0000000..bda414f
--- /dev/null
+++ b/src/main/java/org/swisspush/reststorage/s3/FileReadStream.java
@@ -0,0 +1,274 @@
+package org.swisspush.reststorage.s3;
+
+import io.vertx.core.AsyncResult;
+import io.vertx.core.Context;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.streams.ReadStream;
+import io.vertx.core.streams.impl.InboundBuffer;
+import org.slf4j.Logger;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+
+import static io.vertx.core.file.impl.AsyncFileImpl.DEFAULT_READ_BUFFER_SIZE;
+import static org.slf4j.LoggerFactory.getLogger;
+
+
+public class FileReadStream<T> implements ReadStream<T>, Closeable {
+
+    private static final Logger log = getLogger(FileReadStream.class);
+    private final long expectedSize;
+    private final String path;
+
+    private final ReadableByteChannel ch;
+    private final Vertx vertx;
+    private final Context context;
+
+    private boolean closed;
+    private boolean readInProgress;
+
+    private Handler<Buffer> dataHandler;
+    private Handler<Void> endHandler;
+    private Handler<Throwable> exceptionHandler;
+    private final InboundBuffer<Buffer> queue;
+
+    private final int readBufferSize = DEFAULT_READ_BUFFER_SIZE;
+    private long writtenBytes = 0;
+
+    /**
+     * @param expectedSize Actual file size which is expected to be streamed through that stream
+     *                     in bytes.
+     * @param path         Token printed alongside the logs so when reading logs, we can see which
+     *                     log belongs to which file. A possible candidate is to use the file path
+     *                     but it theoretically can be anything  which  helps  you  to  find  logs
+     *                     related to your observed file.
+     * @param stream       The file (or stream) we wanna observe.
+     */
+    public FileReadStream(Vertx vertx, long expectedSize, String path, InputStream stream) {
+        this.vertx = vertx;
+        this.context = vertx.getOrCreateContext();
+        this.expectedSize = expectedSize;
+        this.path = path;
+        this.ch = Channels.newChannel(stream);
+        this.queue = new InboundBuffer<>(context, 0);
+        queue.handler(buff -> {
+            if (buff.length() > 0) {
+                handleData(buff);
+            } else {
+                handleEnd();
+            }
+        });
+        queue.drainHandler(v -> {
+            doRead();
+        });
+    }
+
+    public void close() {
+        closeInternal(null);
+    }
+
+    public void close(Handler<AsyncResult<Void>> handler) {
+        closeInternal(handler);
+    }
+
+    @Override
+    public ReadStream<T> exceptionHandler(Handler<Throwable> exceptionHandler) {
+        log.trace("exceptionHandler registered for reading '{}'", path);
+        check();
+        this.exceptionHandler = exceptionHandler;
+        return this;
+    }
+
+    @Override
+    public ReadStream<T> handler(Handler handler) {
+        log.trace("handler registered");
+        check();
+        this.dataHandler = handler;
+        if (this.dataHandler != null && !this.closed) {
+            this.doRead();
+        } else {
+            queue.clear();
+        }
+        return this;
+    }
+
+    @Override
+    public ReadStream<T> pause() {
+        log.debug("Pause reading at offset {} for '{}'", writtenBytes, path);
+        check();
+        queue.pause();
+        return this;
+    }
+
+    @Override
+    public ReadStream<T> resume() {
+        log.debug("Resume reading at offset {} for '{}'", writtenBytes, path);
+        check();
+        if (!closed) {
+            queue.resume();
+        }
+        return this;
+    }
+
+    @Override
+    public ReadStream<T> fetch(long amount) {
+        log.debug("fetch amount {}", amount);
+        queue.fetch(amount);
+        return this;
+    }
+
+    @Override
+    public ReadStream<T> endHandler(Handler<Void> endHandler) {
+        log.trace("endHandler registered.");
+        check();
+        this.endHandler = endHandler;
+        log.debug("End handler called ({} bytes remaining) for '{}'", expectedSize - writtenBytes, path);
+        return this;
+    }
+
+    private void doRead() {
+        check();
+        doRead(ByteBuffer.allocate(readBufferSize));
+    }
+
+    private synchronized void doRead(ByteBuffer bb) {
+        if (!readInProgress) {
+            readInProgress = true;
+            Buffer buff = Buffer.buffer(readBufferSize);
+            doRead(buff, 0, bb, writtenBytes, ar -> {
+                if (ar.succeeded()) {
+                    readInProgress = false;
+                    Buffer buffer = ar.result();
+                    writtenBytes += buffer.length();
+                    // Empty buffer represents end of file
+                    if (queue.write(buffer) && buffer.length() > 0) {
+                        doRead(bb);
+                    }
+                } else {
+                    handleException(ar.cause());
+                }
+            });
+        }
+    }
+
+    private void doRead(Buffer writeBuff, int offset, ByteBuffer buff, long position, Handler<AsyncResult<Buffer>> handler) {
+        // ReadableByteChannel doesn't have a completion handler, so we wrap it into
+        // an executeBlocking and use the future there
+        vertx.executeBlocking(future -> {
+            try {
+                Integer bytesRead = ch.read(buff);
+                future.complete(bytesRead);
+            } catch (IOException e) {
+                log.error("Failed to read data from buffer.", e);
+                future.fail(e);
+            }
+
+        }, res -> {
+
+            if (res.failed()) {
+                log.error("Failed to read data from buffer.", res.cause());
+                context.runOnContext((v) -> handler.handle(Future.failedFuture(res.cause())));
+            } else {
+                // Do the completed check
+                Integer bytesRead = (Integer) res.result();
+                if (bytesRead == -1) {
+                    //End of file
+                    context.runOnContext((v) -> {
+                        buff.flip();
+                        writeBuff.setBytes(offset, buff);
+                        buff.compact();
+                        handler.handle(Future.succeededFuture(writeBuff));
+                    });
+                } else if (buff.hasRemaining()) {
+                    long pos = position;
+                    pos += bytesRead;
+                    // resubmit
+                    doRead(writeBuff, offset, buff, pos, handler);
+                } else {
+                    // It's been fully written
+
+                    context.runOnContext((v) -> {
+                        buff.flip();
+                        writeBuff.setBytes(offset, buff);
+                        buff.compact();
+                        handler.handle(Future.succeededFuture(writeBuff));
+                    });
+                }
+            }
+        });
+    }
+
+    private void handleData(Buffer buff) {
+        Handler<Buffer> handler;
+        synchronized (this) {
+            handler = this.dataHandler;
+        }
+        if (handler != null) {
+            checkContext();
+            handler.handle(buff);
+        }
+    }
+
+    private synchronized void handleEnd() {
+        Handler<Void> endHandler;
+        synchronized (this) {
+            dataHandler = null;
+            endHandler = this.endHandler;
+        }
+        if (endHandler != null) {
+            checkContext();
+            endHandler.handle(null);
+        }
+    }
+
+    private void handleException(Throwable t) {
+        if (exceptionHandler != null && t instanceof Exception) {
+            exceptionHandler.handle(t);
+        } else {
+            log.error("Unhandled exception", t);
+        }
+    }
+
+    private void check() {
+        if (this.closed) {
+            throw new IllegalStateException("Inputstream is closed");
+        }
+    }
+
+    public boolean isClosed() {
+        return this.closed;
+    }
+
+    private void checkContext() {
+        if (!vertx.getOrCreateContext().equals(context)) {
+            throw new IllegalStateException("AsyncInputStream must only be used in the context that created it, expected: " + this.context
+                    + " actual " + vertx.getOrCreateContext());
+        }
+    }
+
+    private synchronized void closeInternal(Handler<AsyncResult<Void>> handler) {
+        check();
+        closed = true;
+        doClose(handler);
+    }
+
+    private void doClose(Handler<AsyncResult<Void>> handler) {
+        try {
+            ch.close();
+            if (handler != null) {
+                this.vertx.runOnContext(v -> handler.handle(Future.succeededFuture()));
+            }
+        } catch (IOException e) {
+            if (handler != null) {
+                this.vertx.runOnContext(v -> handler.handle(Future.failedFuture(e)));
+            }
+        }
+    }
+}
diff --git a/src/main/java/org/swisspush/reststorage/s3/FileWriteStream.java b/src/main/java/org/swisspush/reststorage/s3/FileWriteStream.java
new file mode 100644
index 0000000..85bdc70
--- /dev/null
+++ b/src/main/java/org/swisspush/reststorage/s3/FileWriteStream.java
@@ -0,0 +1,115 @@
+package org.swisspush.reststorage.s3;
+
+import io.vertx.core.AsyncResult;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
+import io.vertx.core.Promise;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.streams.WriteStream;
+import org.slf4j.Logger;
+
+import java.io.OutputStream;
+import java.io.IOException;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+public class FileWriteStream implements WriteStream<Buffer> {
+    private static final Logger log = getLogger(FileWriteStream.class);
+    private final OutputStream outputStream;
+    private Handler<Throwable> exceptionHandler;
+    private Handler<Void> drainHandler;
+    private boolean writeQueueFull = false;
+    private int maxQueueSize = 1024;  // Default max queue size, can be modified
+
+    public FileWriteStream(OutputStream outputStream) {
+        this.outputStream = outputStream;
+    }
+
+    @Override
+    public WriteStream<Buffer> exceptionHandler(Handler<Throwable> handler) {
+        this.exceptionHandler = handler;
+        return this;
+    }
+
+    @Override
+    public void write(Buffer buffer, Handler<AsyncResult<Void>> handler) {
+        try {
+            outputStream.write(buffer.getBytes());
+            outputStream.flush();
+            handler.handle(Future.succeededFuture());
+        } catch (IOException e) {
+            if (exceptionHandler != null) {
+                exceptionHandler.handle(e);
+            }
+            log.error("Error writing to stream", e);
+            handler.handle(Future.failedFuture(e));
+        }
+    }
+
+    @Override
+    public Future<Void> write(Buffer buffer) {
+        Promise<Void> promise = Promise.promise();
+        try {
+            outputStream.write(buffer.getBytes());
+            outputStream.flush();
+            promise.complete();
+        } catch (IOException e) {
+            if (exceptionHandler != null) {
+                exceptionHandler.handle(e);
+            }
+            log.error("Error writing to stream", e);
+            promise.fail(e);
+        }
+        return promise.future();
+    }
+
+    @Override
+    public void end(Handler<AsyncResult<Void>> handler) {
+        try {
+            outputStream.flush();
+            outputStream.close();
+            handler.handle(Future.succeededFuture());
+        } catch (IOException e) {
+            if (exceptionHandler != null) {
+                exceptionHandler.handle(e);
+            }
+            log.error("Error when try to end stream", e);
+            handler.handle(Future.failedFuture(e));
+        }
+    }
+
+    @Override
+    public Future<Void> end() {
+        Promise<Void> promise = Promise.promise();
+        try {
+            outputStream.flush();
+            outputStream.close();
+            promise.complete();
+        } catch (IOException e) {
+            if (exceptionHandler != null) {
+                exceptionHandler.handle(e);
+            }
+            log.error("Error when try to end stream", e);
+            promise.fail(e);
+        }
+        return promise.future();
+    }
+
+    @Override
+    public WriteStream<Buffer> setWriteQueueMaxSize(int maxSize) {
+        this.maxQueueSize = maxSize;
+        return this;
+    }
+
+    @Override
+    public boolean writeQueueFull() {
+        // Implement custom logic to manage queue size if needed
+        return writeQueueFull;
+    }
+
+    @Override
+    public WriteStream<Buffer> drainHandler(Handler<Void> handler) {
+        this.drainHandler = handler;
+        return this;
+    }
+}
diff --git a/src/main/java/org/swisspush/reststorage/s3/S3FileSystemDirLister.java b/src/main/java/org/swisspush/reststorage/s3/S3FileSystemDirLister.java
new file mode 100644
index 0000000..a3d2c7d
--- /dev/null
+++ b/src/main/java/org/swisspush/reststorage/s3/S3FileSystemDirLister.java
@@ -0,0 +1,116 @@
+package org.swisspush.reststorage.s3;
+
+import io.vertx.core.Handler;
+import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.swisspush.reststorage.CollectionResource;
+import org.swisspush.reststorage.DocumentResource;
+import org.swisspush.reststorage.Resource;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.stream.Stream;
+
+import static org.swisspush.reststorage.s3.S3FileSystemStorage.S3_PATH_SEPARATOR;
+
+
+/**
+ * This type handles listing of directories in S3.
+ *
+ * <p>Internally it makes use of worker-threads to keep eventloop-thread
+ * responsive.</p>
+ */
+public class S3FileSystemDirLister {
+
+    private static final Logger log = LoggerFactory.getLogger(S3FileSystemDirLister.class);
+    private final Vertx vertx;
+    private final String root;
+
+
+    public S3FileSystemDirLister(Vertx vertx, String root) {
+        this.vertx = vertx;
+        this.root = root;
+    }
+
+    public void handleListingRequest(Path path, final int offset, final int count, final Handler<Resource> handler) {
+        // Delegate work to worker thread from thread pool.
+        log.trace("Delegate to worker pool");
+        final long startTimeMillis = System.currentTimeMillis();
+        vertx.executeBlocking(promise -> {
+            log.trace("Welcome on worker-thread.");
+            listDirBlocking(path, offset, count, (Promise<CollectionResource>) (Promise<?>) promise);
+            log.trace("worker-thread says bye.");
+        }, event -> {
+            log.trace("Welcome back on eventloop-thread.");
+            if (log.isDebugEnabled()) {
+                final long durationMillis = System.currentTimeMillis() - startTimeMillis;
+                log.debug("List directory contents of '{}' took {}ms", path, durationMillis);
+            }
+            if (event.failed()) {
+                log.error("Directory listing failed.", event.cause());
+                final Resource erroneousResource = new Resource() {{
+                    // Set fields according to documentation in Resource class.
+                    name = path.getFileName().toString();
+                    exists = false;
+                    error = rejected = invalid = true;
+                    errorMessage = invalidMessage = event.cause().getMessage();
+                }};
+                handler.handle(erroneousResource);
+            } else {
+                handler.handle((Resource) event.result());
+            }
+        });
+        log.trace("Work delegated.");
+    }
+
+    private void listDirBlocking(Path path, int offset, int count, Promise<CollectionResource> promise) {
+        // Prepare our result.
+        final CollectionResource collection = new CollectionResource() {{
+            items = new ArrayList<>(128);
+        }};
+        try (Stream<Path> source = Files.list(path)) {
+            source.forEach(entry -> {
+                String entryName = entry.getFileName().toString();
+                log.trace("Processing entry '{}'", entryName);
+                // Create resource representing currently processed directory entry.
+                final Resource resource;
+                if (entryName.endsWith(S3_PATH_SEPARATOR)) {
+                    resource = new CollectionResource();
+                    entryName = entryName.replace(S3_PATH_SEPARATOR, "");
+                } else {
+                    resource = new DocumentResource();
+                }
+                resource.name = entryName;
+                collection.items.add(resource);
+            });
+        } catch (IOException e) {
+            promise.fail(e);
+            return;
+        }
+        Collections.sort(collection.items);
+        // Don't know exactly what we do here now. Seems we check 'limit' for a range request.
+        int n = count;
+        if (n == -1) {
+            n = collection.items.size();
+        }
+        // Don't know exactly what we do here. But it seems we evaluate 'start' of a range request.
+        if (offset > -1) {
+            if (offset >= collection.items.size() || (offset + n) >= collection.items.size() || (offset == 0 && n == -1)) {
+                promise.complete(collection);
+            } else {
+                collection.items = collection.items.subList(offset, offset + n);
+                promise.complete(collection);
+            }
+        } else {
+            // TODO: Resolve future
+            //       Previous implementation did nothing here. Why? Should we do something here?
+            //       See: "https://github.com/hiddenalpha/vertx-rest-storage/blob/v2.5.2/src/main/java/org/swisspush/reststorage/FileSystemStorage.java#L77"
+            log.warn("May we should do something here. I've no idea why old implementation did nothing.");
+        }
+    }
+}
diff --git a/src/main/java/org/swisspush/reststorage/s3/S3FileSystemStorage.java b/src/main/java/org/swisspush/reststorage/s3/S3FileSystemStorage.java
new file mode 100644
index 0000000..461769e
--- /dev/null
+++ b/src/main/java/org/swisspush/reststorage/s3/S3FileSystemStorage.java
@@ -0,0 +1,422 @@
+package org.swisspush.reststorage.s3;
+
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.swisspush.reststorage.CollectionResource;
+import org.swisspush.reststorage.DocumentResource;
+import org.swisspush.reststorage.Resource;
+import org.swisspush.reststorage.Storage;
+import org.swisspush.reststorage.exception.RestStorageExceptionFactory;
+import org.swisspush.reststorage.util.LockMode;
+import software.amazon.nio.spi.s3.S3FileSystem;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.nio.file.DirectoryNotEmptyException;
+import java.nio.file.DirectoryStream;
+import java.nio.file.FileSystemAlreadyExistsException;
+import java.nio.file.FileSystemException;
+import java.nio.file.FileSystems;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.LinkOption;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+
+public class S3FileSystemStorage implements Storage {
+
+    private final Path root;
+    private final Vertx vertx;
+    private final RestStorageExceptionFactory exceptionFactory;
+    private final int rootLen;
+    private final S3FileSystemDirLister fileSystemDirLister;
+    private static volatile S3FileSystem fileSystem;
+
+    private final Logger log = LoggerFactory.getLogger(S3FileSystemStorage.class);
+
+    // See: software.amazon.nio.spi.s3.Constants
+    public static final String S3_PATH_SEPARATOR = "/";
+
+    // use only one file system instance
+    private static S3FileSystem getFileSystem(URI uri) {
+        if (fileSystem == null) {
+            synchronized (S3FileSystem.class) {
+                if (fileSystem == null) {
+                    fileSystem = (S3FileSystem) FileSystems.getFileSystem(uri);
+                }
+            }
+        }
+        return fileSystem;
+    }
+
+    public S3FileSystemStorage(Vertx vertx, RestStorageExceptionFactory exceptionFactory, String rootPath,
+                               String awsS3Region, String s3BucketName, String s3AccessKeyId, String s3SecretAccessKey,
+                               boolean useTlsConnection, boolean isLocalS3, String localS3Endpoint, int localS3Port, boolean createBucketIfNotPresentYet) {
+        this.vertx = vertx;
+        this.exceptionFactory = exceptionFactory;
+        Objects.requireNonNull(s3BucketName, "BucketName must not be null");
+        Objects.requireNonNull(awsS3Region, "Region must not be null");
+        System.setProperty("aws.region", awsS3Region);
+
+        if (isLocalS3) {
+            // local S3, AWS SDK requires these two properties to be set
+            System.setProperty("aws.accessKeyId", "local");
+            System.setProperty("aws.secretAccessKey", "local");
+
+            String credentials = "";
+            if (StringUtils.isNotEmpty(s3AccessKeyId) && StringUtils.isNotEmpty(s3SecretAccessKey)) {
+                credentials = s3AccessKeyId + ":" + s3SecretAccessKey + "@";
+            }
+            String port = "";
+            if (localS3Port > 0) {
+                port = ":" + localS3Port;
+            }
+            // s3x://[key:secret@]endpoint[:port]/bucket
+            s3BucketName = "s3x://" + credentials + localS3Endpoint + port + "/" + s3BucketName;
+            if (!useTlsConnection) {
+                System.setProperty("s3.spi.endpoint-protocol", "http");
+            }
+        } else {
+            // AWS S3
+            Objects.requireNonNull(s3AccessKeyId, "AccessKeyId must not be null");
+            Objects.requireNonNull(s3SecretAccessKey, "SecretAccessKey must not be null");
+            System.setProperty("aws.accessKeyId", s3AccessKeyId);
+            System.setProperty("aws.secretAccessKey", s3SecretAccessKey);
+            s3BucketName = "s3://" + s3BucketName;
+        }
+
+        var uri = URI.create(s3BucketName);
+
+        if (createBucketIfNotPresentYet) {
+            try (var fs = FileSystems.newFileSystem(uri,
+                    Map.of("locationConstraint", awsS3Region))) {
+                log.info("Bucket created: " + fs.toString());
+            } catch (FileSystemAlreadyExistsException e) {
+                log.info("Bucket " + s3BucketName + " already exists: ", e);
+            } catch (IOException e) {
+                log.error("Failed to create bucket " + s3BucketName, e);
+            }
+        }
+
+        fileSystem = getFileSystem(uri);
+        root = fileSystem.getPath(rootPath);
+
+        try (var pathStream = Files.walk(root)) {
+            pathStream.forEach(System.out::println);
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+
+        this.fileSystemDirLister = new S3FileSystemDirLister(vertx, root.toString());
+        // Unify format for simpler work.
+        String tmpRoot;
+        tmpRoot = root.toString();
+
+        // Cache string length of root without trailing slashes
+        int rootLen;
+        for (rootLen = tmpRoot.length() - 1; tmpRoot.charAt(rootLen) == '/'; --rootLen);
+        this.rootLen = rootLen;
+    }
+
+    @Override
+    public Optional<Float> getCurrentMemoryUsage() {
+        throw new UnsupportedOperationException("Method 'getCurrentMemoryUsage' not supported in S3FileSystemStorage");
+    }
+
+    @Override
+    public void get(String path, String etag, final int offset, final int count, final Handler<Resource> handler) {
+        final Path fullDirPath = canonicalize(path, true);
+        final Path fullFilePath = canonicalize(path, false);
+        log.debug("GET {}", path);
+        if (Files.isRegularFile(fullFilePath, LinkOption.NOFOLLOW_LINKS)) {
+            log.debug("Open file '{}' ({})", path, fullFilePath);
+            DocumentResource d = new DocumentResource();
+            // DON'T close it with try or finally, async code
+            try {
+                final InputStream inputStream = Files.newInputStream(fullFilePath);
+                d.length = Files.size(fullFilePath);
+                log.debug("Successfully opened '{}' which is {} bytes in size.", path, d.length);
+
+                final FileReadStream<io.vertx.core.buffer.Buffer> readStream = new FileReadStream<>(vertx, d.length, path, inputStream);
+                d.readStream = readStream;
+
+                final Runnable cleanUp = () -> {
+                    if (!readStream.isClosed()){
+                    readStream.close();
+                    }
+                    try {
+                        inputStream.close();
+                    } catch (IOException e) {
+                        log.debug("Failed to close input stream", e);
+                    }
+                };
+                d.closeHandler = v -> {
+                    log.debug("Resource got closed. Close file now '{}'", path);
+                    cleanUp.run();
+                };
+                d.addErrorHandler(event -> {
+                    log.error("Get resource failed.", exceptionFactory.newException("Close file now '" + path + "'", event));
+                    cleanUp.run();
+                });
+            } catch (IOException e) {
+                log.warn("Failed to open '{}' for read", path, e);
+                d.error = true;
+                d.errorMessage = e.getMessage();
+            }
+            handler.handle(d);
+            return;
+        }
+
+        if (Files.isDirectory(fullDirPath)) {
+            log.debug("Delegate directory listing of '{}'", path);
+            fileSystemDirLister.handleListingRequest(fullDirPath, offset, count, handler);
+        } else {
+            log.debug("No such file or directory '{}' ({})", path, fullDirPath);
+            Resource r = new Resource();
+            r.exists = false;
+            handler.handle(r);
+        }
+    }
+
+    @Override
+    public void put(String path, String etag, boolean merge, long expire, final Handler<Resource> handler) {
+        put(path, etag, merge, expire, "", LockMode.SILENT, 0, handler);
+    }
+
+    @Override
+    public void put(String path, String etag, boolean merge, long expire, String lockOwner, LockMode lockMode, long lockExpire, Handler<Resource> handler) {
+        final Path fullPath = canonicalize(path, false);
+        if (Files.exists(fullPath, LinkOption.NOFOLLOW_LINKS)) {
+            if (Files.isDirectory(fullPath)) {
+                CollectionResource c = new CollectionResource();
+                handler.handle(c);
+            } else if (Files.isRegularFile(fullPath, LinkOption.NOFOLLOW_LINKS)) {
+                putFile(handler, fullPath);
+            } else {
+                Resource r = new Resource();
+                r.exists = false;
+                handler.handle(r);
+            }
+        } else {
+            if (Files.exists(fullPath)) {
+                putFile(handler, fullPath);
+            } else {
+                try {
+                    Files.createDirectory(fullPath.getParent());
+                    putFile(handler, fullPath);
+                } catch (IOException e) {
+                    log.error("Failed to create directory '{}'", path);
+                }
+            }
+        }
+    }
+
+    @Override
+    public void put(String path, String etag, boolean merge, long expire, String lockOwner, LockMode lockMode, long lockExpire, boolean storeCompressed, Handler<Resource> handler) {
+        if (storeCompressed) {
+            log.warn("PUT with storeCompressed option is not yet implemented in file system storage. Ignoring storeCompressed option value");
+        }
+        put(path, etag, merge, expire, "", LockMode.SILENT, 0, handler);
+    }
+
+    private void putFile(final Handler<Resource> handler, final Path fullPath) {
+        try {
+            // DON'T close it in finally block, async code
+            final OutputStream outputStream = Files.newOutputStream(fullPath);
+            final DocumentResource d = new DocumentResource();
+
+            FileWriteStream fileWriteStream = new FileWriteStream(outputStream);
+            d.writeStream = fileWriteStream;
+
+            final Runnable cleanUp = () -> {
+
+                try {
+                    outputStream.close();
+                } catch (IOException e) {
+                    log.error("Failed to close output stream:", e);
+                }
+                fileWriteStream.end();
+            };
+
+            d.closeHandler = event -> {
+                cleanUp.run();
+                d.endHandler.handle(null);
+            };
+            d.addErrorHandler(err -> {
+                log.error("Put file failed:", err);
+                cleanUp.run();
+            });
+            handler.handle(d);
+        } catch (IOException e) {
+            log.error("Failed to put file:", e);
+        }
+    }
+
+    @Override
+    public void delete(String path, String lockOwner, LockMode lockMode, long lockExpire, boolean confirmCollectionDelete,
+                       boolean deleteRecursive, final Handler<Resource> handler) {
+
+        // at this stage I don't know what of this path point to
+        final Path fullFilePath = canonicalize(path, false);
+        final Path fullDirPath = canonicalize(path, true);
+
+        Resource resource = new Resource();
+        boolean deleteRecursiveInFileSystem = true;
+        if (confirmCollectionDelete && !deleteRecursive) {
+            deleteRecursiveInFileSystem = false;
+        }
+        boolean finalDeleteRecursiveInFileSystem = deleteRecursiveInFileSystem;
+
+        // try to check as a file
+        boolean exists = Files.exists(fullFilePath, LinkOption.NOFOLLOW_LINKS);
+        Path fullPath = fullFilePath;
+
+        if (!exists) {
+            // not exists, maybe is a directory
+            exists = Files.exists(fullDirPath, LinkOption.NOFOLLOW_LINKS);
+            if (exists) {
+                // so it is a directory
+                fullPath = fullDirPath;
+            }
+        }
+
+        if (exists) {
+            try {
+                deleteRecursive(fullPath, finalDeleteRecursiveInFileSystem);
+                deleteEmptyParentDirs(fullPath.getParent());
+            } catch (IOException e) {
+                if (e.getCause() != null && e.getCause() instanceof DirectoryNotEmptyException) {
+                    resource.error = true;
+                    resource.errorMessage = "directory not empty. Use recursive=true parameter to delete";
+                } else {
+                    resource.exists = false;
+                }
+            }
+            handler.handle(resource);
+        } else {
+            Resource r = new Resource();
+            r.exists = false;
+            handler.handle(r);
+        }
+    }
+
+    @Override
+    public void cleanup(Handler<DocumentResource> handler, String cleanupResourcesAmount) {
+        // nothing to do here
+    }
+
+    @Override
+    public void storageExpand(String path, String etag, List<String> subResources, Handler<Resource> handler) {
+        throw new UnsupportedOperationException("Method 'storageExpand' not supported in S3FileSystemStorage");
+    }
+
+    /**
+     * Deletes all empty parent directories starting at specified directory.
+     *
+     * @param path Most deep (virtual) directory to start bubbling up deletion of empty
+     *             directories.
+     */
+    private void deleteEmptyParentDirs(Path path) {
+        // Analyze if we reached root.
+        int pathLen;
+        // Evaluate length of current path excluding trailing slashes by searching
+        // last non-slash (backslash of course on windows).
+        for (pathLen = path.toString().length() - 1; path.toString().charAt(pathLen) == File.separatorChar; --pathLen) ;
+        if (rootLen == pathLen) {
+            // We do NOT want to delete our virtual root even it is empty :)
+            log.debug("Stop deletion here to keep virtual root '{}'.", root);
+            return;
+        }
+        log.debug("Delete directory if empty '{}'.", path);
+
+        if (!isDirEmpty(path)) {
+            log.debug("Directory is NOT empty '{}'.", path);
+            return;
+        }
+
+        try {
+            if (Files.deleteIfExists(path)) {
+                // Bubbling up to parent.
+                final Path parentPath = path.getParent();
+                // HINT 1: We go recursive here!
+                // HINT 2: When debugging stack traces keep in mind this recursion occurs
+                //         asynchronous and therefore is not really a recursion :)
+                deleteEmptyParentDirs(parentPath);
+            } else {
+                throw new FileNotFoundException();
+            }
+        } catch (IOException cause) {
+            if (cause instanceof FileSystemException && cause.getCause() instanceof DirectoryNotEmptyException) {
+                // Failed to delete directory because it's not empty. Therefore we must not
+                // delete it at all and we're done now.
+                log.debug("Directory '{}' not empty. Stop bubbling deleting dirs.", path);
+            } else if (cause instanceof FileSystemException && cause.getCause() instanceof NoSuchFileException) {
+                // Somehow a caller requested to delete a directory which seems not to exist.
+                // This should never be the case theoretically. (except maybe some race
+                // conditions?)
+                log.warn("Ignored to delete non-existing dir '{}'.", path);
+            } else {
+                // This case should not happen. At least up to now i've no idea of a valid
+                // scenario for this one.
+                log.error("Unexpected error while deleting empty directories.", exceptionFactory.newException(cause));
+            }
+        }
+    }
+
+    public Path canonicalize(String path, boolean isDir) {
+        if (path.startsWith(S3_PATH_SEPARATOR)) {
+            path = path.substring(1);
+        }
+
+        if (!path.endsWith(S3_PATH_SEPARATOR) && isDir) {
+            path = path + S3_PATH_SEPARATOR;
+        }
+        return root.resolve(path);
+    }
+
+    private void deleteRecursive(Path path, boolean recursive) throws IOException {
+        if (recursive) {
+            Files.walkFileTree(path, new SimpleFileVisitor<>() {
+                public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
+                    Files.delete(file);
+                    return FileVisitResult.CONTINUE;
+                }
+
+                public FileVisitResult postVisitDirectory(Path dir, IOException e) throws IOException {
+                    if (e == null) {
+                        Files.delete(dir);
+                        return FileVisitResult.CONTINUE;
+                    } else {
+                        throw e;
+                    }
+                }
+            });
+        } else {
+            Files.delete(path);
+        }
+    }
+
+    private boolean isDirEmpty(final Path path) {
+        try (DirectoryStream<Path> dirStream = Files.newDirectoryStream(path)) {
+            return !dirStream.iterator().hasNext();
+        } catch (IOException e) {
+            log.debug("Error to detects is directory empty or not '{}'.", path, exceptionFactory.newException(e));
+            return false;
+        }
+    }
+}
diff --git a/src/main/java/org/swisspush/reststorage/util/ModuleConfiguration.java b/src/main/java/org/swisspush/reststorage/util/ModuleConfiguration.java
index 2e13921..ee3b0b3 100644
--- a/src/main/java/org/swisspush/reststorage/util/ModuleConfiguration.java
+++ b/src/main/java/org/swisspush/reststorage/util/ModuleConfiguration.java
@@ -21,7 +21,7 @@ public class ModuleConfiguration {
     private static final Logger log = LoggerFactory.getLogger(ModuleConfiguration.class);
 
     public enum StorageType {
-        filesystem, redis
+        filesystem, redis, s3
     }
 
     private String root = ".";
@@ -70,6 +70,15 @@ public enum StorageType {
     private int maxRedisWaitingHandlers = 2048;
     private int maxStorageExpandSubresources = 1000;
 
+    private String s3BucketName = null;
+    private String awsS3Region = null;
+    private String s3AccessKeyId = null;
+    private String s3SecretAccessKey = null;
+    private boolean s3UseTlsConnection = true;
+    private boolean createBucketIfNotPresentYet = false;
+    private boolean localS3 = false;
+    private String localS3Endpoint = null;
+    private int localS3Port = 0;
 
     public ModuleConfiguration root(String root) {
         this.root = root;
@@ -292,6 +301,51 @@ public ModuleConfiguration maxStorageExpandSubresources(int maxStorageExpandSubr
         return this;
     }
 
+    public ModuleConfiguration awsS3Region(String awsS3Region) {
+        this.awsS3Region = awsS3Region;
+        return this;
+    }
+
+    public ModuleConfiguration s3BucketName(String awsS3BucketName) {
+        this.s3BucketName = awsS3BucketName;
+        return this;
+    }
+
+    public ModuleConfiguration s3AccessKeyId(String awsS3AccessKeyId) {
+        this.s3AccessKeyId = awsS3AccessKeyId;
+        return this;
+    }
+
+    public ModuleConfiguration s3SecretAccessKey(String awsS3SecretAccessKey) {
+        this.s3SecretAccessKey = awsS3SecretAccessKey;
+        return this;
+    }
+
+    public ModuleConfiguration s3UseTlsConnection(boolean s3UseTlsConnection) {
+        this.s3UseTlsConnection = s3UseTlsConnection;
+        return this;
+    }
+
+    public ModuleConfiguration localS3Endpoint(String s3Endpoint) {
+        this.localS3Endpoint = s3Endpoint;
+        return this;
+    }
+
+    public ModuleConfiguration localS3Port(int s3Port) {
+        this.localS3Port = s3Port;
+        return this;
+    }
+
+    public ModuleConfiguration createBucketIfNotPresentYet(boolean createBucketIfNotExist) {
+        this.createBucketIfNotPresentYet = createBucketIfNotExist;
+        return this;
+    }
+
+    public ModuleConfiguration localS3(boolean localS3) {
+        this.localS3 = localS3;
+        return this;
+    }
+
     public String getRoot() {
         return root;
     }
@@ -465,6 +519,42 @@ public int getMaxStorageExpandSubresources() {
         return maxStorageExpandSubresources;
     }
 
+    public String getAwsS3Region() {
+        return awsS3Region;
+    }
+
+    public String getS3BucketName() {
+        return s3BucketName;
+    }
+
+    public String getS3AccessKeyId() {
+        return s3AccessKeyId;
+    }
+
+    public String getS3SecretAccessKey() {
+        return s3SecretAccessKey;
+    }
+
+    public boolean getS3UseTlsConnection() {
+        return s3UseTlsConnection;
+    }
+
+    public String getLocalS3Endpoint() {
+        return localS3Endpoint;
+    }
+
+    public int getLocalS3Port() {
+        return localS3Port;
+    }
+
+    public boolean getCreateBucketIfNotPresentYet() {
+        return createBucketIfNotPresentYet;
+    }
+
+    public boolean isLocalS3() {
+        return localS3;
+    }
+
     public JsonObject asJsonObject() {
         return JsonObject.mapFrom(this);
     }