@Override
    public <T> void getBinary(final String cacheKey, final Class<T> type,
            final IAsyncResultHandler<ISignalReadStream<T>> handler) {
        try {
            GetResponse response = getClient().get(new GetRequest(getFullIndexName()).id(cacheKey), RequestOptions.DEFAULT);

            // Did the GET succeed?  If not, return null.
            if (!response.isExists()) {
                handler.handle(AsyncResultImpl.create((ISignalReadStream<T>) null));
                return;
            }

            // Is the cache entry expired?  If so return null.
            String sourceAsString = response.getSourceAsString();
            CacheEntry cacheEntry = JSON_MAPPER.readValue(sourceAsString, CacheEntry.class);
            if (System.currentTimeMillis() > cacheEntry.getExpiresOn() ) {
                // Cache item has expired.  Return null instead of the cached data.
                handler.handle(AsyncResultImpl.create((ISignalReadStream<T>) null));
                return;
            }

            try {
                final T head = (T) JSON_MAPPER.reader(type).readValue(cacheEntry.getHead());
                String b64Data = cacheEntry.getData();
                final IApimanBuffer data = bufferFactory.createBuffer(Base64.decodeBase64(b64Data));
                ISignalReadStream<T> rval = new ISignalReadStream<T>() {
                    IAsyncHandler<IApimanBuffer> bodyHandler;
                    IAsyncHandler<Void> endHandler;
                    boolean finished = false;
                    boolean aborted = false;

                    @Override
                    public void bodyHandler(IAsyncHandler<IApimanBuffer> bodyHandler) {
                        this.bodyHandler = bodyHandler;
                    }

                    @Override
                    public void endHandler(IAsyncHandler<Void> endHandler) {
                        this.endHandler = endHandler;
                    }

                    @Override
                    public T getHead() {
                        return head;
                    }

                    @Override
                    public boolean isFinished() {
                        return finished;
                    }

                    @Override
                    public void abort(Throwable t) {
                        finished = true;
                        aborted = true;
                    }

                    @Override
                    public void transmit() {
                        if (!aborted) {
                            bodyHandler.handle(data);
                            endHandler.handle(null);
                        }
                        finished = true;
                    }
                };
                handler.handle(AsyncResultImpl.create(rval));
            } catch (Throwable e) {
                LOGGER.error(e, "Error attempting to stream cached binary on key {0}", cacheKey);
                handler.handle(AsyncResultImpl.create((ISignalReadStream<T>) null));
            }
        } catch (Throwable e) {
            LOGGER.error(e, "Error attempting to GET cached binary on key {0}", cacheKey);
            handler.handle(AsyncResultImpl.create((ISignalReadStream<T>) null));
        }
    }
@see io.apiman.gateway.engine.components.ICacheStoreComponent#getBinary(java.lang.String, java.lang.Class, io.apiman.gateway.engine.async.IAsyncResultHandler)
gateway/engine/es/src/main/java/io/apiman/gateway/engine/es/EsCacheStoreComponent.java