@Override
    public void store(final Document document, Reader reader) {

        final String documentId = document.getId();
        LOGGER.info(String.format("store document '%s'", documentId));

        KeyHolder keyHolder = new GeneratedKeyHolder();

        delete(documentId);

        LOGGER.debug(String.format("persist document '%s'", document.getId()));
        jdbcTemplate.update(
                new PreparedStatementCreator() {
                    public PreparedStatement createPreparedStatement(Connection connection) throws SQLException {
                        PreparedStatement ps =
                                connection.prepareStatement("INSERT INTO `documents` ( `identifier`, `hash`, `url` ) VALUES ( ?, ?, ? )", new String[]{"id", "hash", "url"});

                        ps.setString(1, documentId);
                        ps.setString(2, document.getHash());
                        ps.setString(3, document.getUrl());
                        return ps;
                    }
                },
                keyHolder);

        DefaultTransactionDefinition def = new DefaultTransactionDefinition();
        def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);

        txManager.setDataSource(jdbcTemplate.getDataSource());
        TransactionStatus status = txManager.getTransaction(def);

//        ChunkProvider provider = new ChunkProvider(reader, chunkSize);
        try {
            int chunkId = 0;

            if (useCompression) {
                PackedChunkProvider provider = new PackedChunkProvider(reader, chunkSize);
//                long totalBytes = 0;

                while (provider.hasNextChunk()) {
                    PackedChunk chunk = provider.nextChunk();
                    if (chunk == null) {
                        break;
                    }
//                totalBytes += chunk.getLength();
//                storeChunk(chunkId++, keyHolder.getKey().intValue(), chunk, charset);
//                    storeChunk(chunkId++, keyHolder.getKey().intValue(), chunk.getPackedPayload(), charset);
                storeChunk(chunkId++, keyHolder.getKey().intValue(), ((PackedChunk)chunk).getPackedPayload(), charset);
                }
//            LOGGER.info(String.format("compression for document '%s': %s%%", documentId, ((double) totalBytes) / (chunkId * chunkSize)));

            } else {

                ChunkProvider provider = new ChunkProvider(reader, chunkSize);
                while (provider.hasNextChunk()) {
                    Chunk chunk = provider.nextChunk();
                    if (chunk == null) {
                        break;
                    }
                    storeChunk(chunkId++, keyHolder.getKey().intValue(), chunk.getPayload().getBytes(), charset);
                }
            }


            LOGGER.info(String.format("commit %s chunks for document '%s'", chunkId, documentId));

            txManager.commit(status);

        } catch (IOException e) {
            txManager.rollback(status);
            LOGGER.error(String.format("storing chunks for doc %s failed: %s", documentId, e.getMessage()));
            LOGGER.debug(e);

        } finally {
            try {
                reader.close();
            } catch (IOException e) {
                //
            }
        }
    }