/**
 * Copyright 2016 Palantir Technologies
 *
 * Licensed under the BSD-3 License (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://opensource.org/licenses/BSD-3-Clause
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.palantir.atlasdb.performance.benchmarks.table;

import org.openjdk.jmh.annotations.Level;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.TearDown;

import com.palantir.atlasdb.keyvalue.api.KeyValueService;
import com.palantir.atlasdb.keyvalue.api.TableReference;
import com.palantir.atlasdb.performance.backend.AtlasDbServicesConnector;
import com.palantir.atlasdb.performance.benchmarks.Benchmarks;
import com.palantir.atlasdb.services.AtlasDbServices;
import com.palantir.atlasdb.transaction.api.TransactionManager;

@State(Scope.Benchmark)
public class StreamingTable {
    private AtlasDbServicesConnector connector;
    private AtlasDbServices services;

    public TransactionManager getTransactionManager() {
        return services.getTransactionManager();
    }

    public KeyValueService getKvs() {
        return services.getKeyValueService();
    }

    public TableReference getTableRef() {
        return Tables.TABLE_REF;
    }

    @TearDown(Level.Trial)
    public void cleanup() throws Exception {
        this.connector.close();
    }

    @Setup(Level.Trial)
    public void setup(AtlasDbServicesConnector conn) {
        this.connector = conn;
        services = conn.connect();
        if (!services.getKeyValueService().getAllTableNames().contains(getTableRef())) {
            Benchmarks.createTableWithStreaming(getKvs(), getTableRef(), Tables.ROW_COMPONENT, Tables.COLUMN_NAME);
            setupData();
        }
    }

    private void setupData() {
        // TODO store a large amount of data so we can stream it
    }
}

@@ -642,16 +642,107 @@
 table;%0A%0A
+import java.io.ByteArrayInputStream;%0Aimport java.io.InputStream;%0Aimport java.util.Random;%0A%0A
 import o
@@ -947,16 +947,111 @@
 rDown;%0A%0A
+import com.google.common.collect.ImmutableMultimap;%0Aimport com.google.common.collect.Multimap;%0A
 import c
@@ -1326,30 +1326,368 @@
 sdb.
-services.AtlasDbServic
+performance.schema.generated.KeyValueTable;%0Aimport com.palantir.atlasdb.performance.schema.generated.StreamTestTableFactory;%0Aimport com.palantir.atlasdb.performance.schema.generated.ValueStreamStore;%0Aimport com.palantir.atlasdb.services.AtlasDbServices;%0Aimport com.palantir.atlasdb.table.api.ColumnValue;%0Aimport com.palantir.atlasdb.table.generation.ColumnValu
 es;%0A
@@ -1749,16 +1749,64 @@
 Manager;
+%0Aimport com.palantir.common.persist.Persistable;
 %0A%0A@State
@@ -1853,16 +1853,77 @@
 Table %7B%0A
+    private Random random = new Random(Tables.RANDOM_SEED);%0A%0A
     priv
@@ -2832,64 +2832,947 @@
     
-// TODO store a large amount of data so we can stream it
+StreamTestTableFactory tableFactory = StreamTestTableFactory.of();%0A        ValueStreamStore streamTestStreamStore = ValueStreamStore.of(%0A                getTransactionManager(),%0A                tableFactory%0A        );%0A        byte%5B%5D data = new byte%5B6_000_000%5D;%0A%0A        random.nextBytes(data);%0A%0A        InputStream inputStream = new ByteArrayInputStream(data);%0A        long streamId = streamTestStreamStore.storeStream(inputStream).getLhSide();%0A%0A        getTransactionManager().runTaskThrowOnConflict(txn -%3E %7B%0A            KeyValueTable table = tableFactory.getKeyValueTable(txn);%0A            KeyValueTable.KeyValueRow row = KeyValueTable.KeyValueRow.of(%22row%22);%0A            KeyValueTable.StreamId id = KeyValueTable.StreamId.of(streamId);%0A            Multimap%3CPersistable, ColumnValue%3C?%3E%3E rows = ImmutableMultimap.of(row, id);%0A%0A            txn.put(table.getTableRef(), ColumnValues.toCellValues(rows));%0A            return null;%0A        %7D);%0A
 %0A   

Add a single, 6MB, row in StreamingTable.