// Copyright 2015 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.schema.stream;

import java.io.ByteArrayInputStream;

import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import com.palantir.atlasdb.AtlasDbTestCase;
import com.palantir.atlasdb.encoding.PtBytes;
import com.palantir.atlasdb.schema.stream.generated.StreamTestStreamStore;
import com.palantir.atlasdb.schema.stream.generated.StreamTestTableFactory;
import com.palantir.atlasdb.stream.PersistentStreamStore;
import com.palantir.atlasdb.transaction.api.Transaction;
import com.palantir.atlasdb.transaction.api.TransactionTask;
import com.palantir.util.crypto.Sha256Hash;

public class StreamTest extends AtlasDbTestCase {

    @Before
    public void createSchema() {
        StreamTestSchema.getSchema().deleteTablesAndIndexes(keyValueService);
        StreamTestSchema.getSchema().createTablesAndIndexes(keyValueService);
    }

    @Test
    public void testAddDelete() throws Exception {
        txManager.runTaskWithRetry(new TransactionTask<Void, Exception>() {
            @Override
            public Void execute(Transaction t) throws Exception {
                PersistentStreamStore store = StreamTestStreamStore.of(txManager, StreamTestTableFactory.of());
                byte[] data = PtBytes.toBytes("streamed");
                Sha256Hash hash = Sha256Hash.computeHash(data);
                byte[] reference = "ref".getBytes();
                long streamId = store.getByHashOrStoreStreamAndMarkAsUsed(t, hash, new ByteArrayInputStream(data), reference);
                Assert.assertEquals(data.length, store.loadStream(t, streamId).read(data, 0, data.length));
//                store.markStreamsAsUsed(ImmutableMap.of(streamId, reference));
//                store.removeStreamsAsUsed(ImmutableMap.of(1L, reference));
                try {
                    store.loadStream(t, 1L).read(data, 0, data.length);
                } catch (IllegalArgumentException e) {
                    // expected
                    return null;
                }
                Assert.fail();
                return null;
            }
        });
    }
}

@@ -1522,16 +1522,95 @@
 %0A       
+ final byte%5B%5D data = PtBytes.toBytes(%22streamed%22);%0A        final long streamId =
  txManag
@@ -1649,20 +1649,20 @@
 ionTask%3C
-Void
+Long
 , Except
@@ -1715,12 +1715,12 @@
 lic 
-Void
+Long
  exe
@@ -2177,116 +2177,8 @@
 e);%0A
-                Assert.assertEquals(data.length, store.loadStream(t, streamId).read(data, 0, data.length));%0A
 //  
@@ -2590,24 +2590,467 @@
 ert.fail();%0A
+                return streamId;%0A            %7D%0A        %7D);%0A        txManager.runTaskWithRetry(new TransactionTask%3CVoid, Exception%3E() %7B%0A            @Override%0A            public Void execute(Transaction t) throws Exception %7B%0A                PersistentStreamStore store = StreamTestStreamStore.of(txManager, StreamTestTableFactory.of());%0A                Assert.assertEquals(data.length, store.loadStream(t, streamId).read(data, 0, data.length));%0A
             

fix stream test