<commit_before>/*
 * Copyright (c) 2008-2013, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * 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.hazelcast.queue;

import com.hazelcast.config.Config;
import com.hazelcast.config.QueueConfig;
import com.hazelcast.config.QueueStoreConfig;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IQueue;
import com.hazelcast.core.QueueStore;
import com.hazelcast.test.HazelcastTestSupport;
import com.hazelcast.test.HazelcastSerialClassRunner;
import com.hazelcast.test.TestHazelcastInstanceFactory;
import com.hazelcast.test.annotation.QuickTest;
import org.junit.Assert;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.Assert.assertTrue;

/**
 * User: sancar
 * Date: 2/22/13
 * Time: 2:20 PM
 */
@RunWith(HazelcastSerialClassRunner.class)
@Category(QuickTest.class)
public class QueueStoreTest extends HazelcastTestSupport {

    @Test
    public void testQueueStoreLoadMoreThanMaxSize() {
        Config config = new Config();
        int maxSize = 2000;
        QueueConfig queueConfig = config.getQueueConfig("testQueueStore");
        queueConfig.setMaxSize(maxSize);
        TestQueueStore queueStore = new TestQueueStore();
        QueueStoreConfig queueStoreConfig = new QueueStoreConfig();
        queueStoreConfig.setStoreImplementation(queueStore);
        queueConfig.setQueueStoreConfig(queueStoreConfig);

        TestHazelcastInstanceFactory factory = createHazelcastInstanceFactory(1);
        HazelcastInstance instance = factory.newHazelcastInstance(config);

        for (int i = 0; i < maxSize * 2; i++) {
            queueStore.store.put((long) i, i);
        }

        IQueue<Object> queue = instance.getQueue("testQueueStore");
        Assert.assertEquals("Queue Size should be equal to max size", maxSize, queue.size());
    }

    @Test
    public void testQueueStore() {
        Config config = new Config();
        int maxSize = 2000;
        QueueConfig queueConfig = config.getQueueConfig("testQueueStore");
        queueConfig.setMaxSize(maxSize);
        TestQueueStore queueStore = new TestQueueStore(1000,0,2000,0,0,0,1);

        QueueStoreConfig queueStoreConfig = new QueueStoreConfig();
        queueStoreConfig.setStoreImplementation(queueStore);
        queueConfig.setQueueStoreConfig(queueStoreConfig);

        TestHazelcastInstanceFactory factory = createHazelcastInstanceFactory(2);
        HazelcastInstance instance = factory.newHazelcastInstance(config);

        for (int i = 0; i < maxSize / 2; i++) {
            queueStore.store.put((long) i, i);
        }

        IQueue<Object> queue = instance.getQueue("testQueueStore");

        for (int i = 0; i < maxSize / 2; i++) {
            queue.offer(i + maxSize / 2);
        }

        instance.getLifecycleService().shutdown();
        HazelcastInstance instance2 = factory.newHazelcastInstance(config);

        IQueue<Object> queue2 = instance2.getQueue("testQueueStore");
        Assert.assertEquals(maxSize,queue2.size());


        Assert.assertEquals(maxSize,queueStore.store.size());
        for (int i = 0; i < maxSize; i++) {
            Assert.assertEquals(i,queue2.poll());
        }

        try {
            queueStore.assertAwait(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static class TestQueueStore implements QueueStore {

        final Map<Long, Integer> store = new LinkedHashMap<Long, Integer>();
        final CountDownLatch latchStore;
        final CountDownLatch latchStoreAll;
        final CountDownLatch latchDelete;
        final CountDownLatch latchDeleteAll;
        final CountDownLatch latchLoad;
        final CountDownLatch latchLoadAllKeys;
        final CountDownLatch latchLoadAll;
        final AtomicInteger callCount = new AtomicInteger();
        final AtomicInteger initCount = new AtomicInteger();
        final AtomicInteger destroyCount = new AtomicInteger();

        private boolean loadAllKeys = true;

        public TestQueueStore() {
            this(0, 0, 0, 0, 0, 0);
        }

        public TestQueueStore(int expectedStore, int expectedDelete, int expectedLoad) {<commit_msg>added test<commit_after>/*
 * Copyright (c) 2008-2013, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * 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.hazelcast.queue;

import com.hazelcast.config.Config;
import com.hazelcast.config.QueueConfig;
import com.hazelcast.config.QueueStoreConfig;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IQueue;
import com.hazelcast.core.QueueStore;
import com.hazelcast.core.TransactionalQueue;
import com.hazelcast.test.HazelcastSerialClassRunner;
import com.hazelcast.test.HazelcastTestSupport;
import com.hazelcast.test.TestHazelcastInstanceFactory;
import com.hazelcast.test.annotation.QuickTest;
import com.hazelcast.transaction.TransactionContext;
import org.junit.Assert;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;

import java.io.Serializable;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

/**
 * User: sancar
 * Date: 2/22/13
 * Time: 2:20 PM
 */
@RunWith(HazelcastSerialClassRunner.class)
@Category(QuickTest.class)
public class QueueStoreTest extends HazelcastTestSupport {

    @Test
    public void testQueueStoreLoadMoreThanMaxSize() {
        Config config = new Config();
        int maxSize = 2000;
        QueueConfig queueConfig = config.getQueueConfig("testQueueStore");
        queueConfig.setMaxSize(maxSize);
        TestQueueStore queueStore = new TestQueueStore();
        QueueStoreConfig queueStoreConfig = new QueueStoreConfig();
        queueStoreConfig.setStoreImplementation(queueStore);
        queueConfig.setQueueStoreConfig(queueStoreConfig);

        TestHazelcastInstanceFactory factory = createHazelcastInstanceFactory(1);
        HazelcastInstance instance = factory.newHazelcastInstance(config);

        for (int i = 0; i < maxSize * 2; i++) {
            queueStore.store.put((long) i, i);
        }

        IQueue<Object> queue = instance.getQueue("testQueueStore");
        Assert.assertEquals("Queue Size should be equal to max size", maxSize, queue.size());
    }

    @Test
    public void testIssue1401QueueStoreWithTxnPoll() {
        final MyQueueStore store = new MyQueueStore();
        final Config config = new Config();
        final QueueConfig qConfig = config.getQueueConfig("test");
        qConfig.setMaxSize(10);
        final QueueStoreConfig queueStoreConfig = new QueueStoreConfig();
        queueStoreConfig.setStoreImplementation(store);
        queueStoreConfig.setEnabled(true);
        queueStoreConfig.setProperty("binary", "false");
        queueStoreConfig.setProperty("memory-limit", "0");
        queueStoreConfig.setProperty("bulk-load", "100");
        qConfig.setQueueStoreConfig(queueStoreConfig);

        TestHazelcastInstanceFactory factory = createHazelcastInstanceFactory(1);
        HazelcastInstance instance = factory.newHazelcastInstance(config);

        for (int i = 0; i < 10; i++) {
            TransactionContext context = instance.newTransactionContext();
            context.beginTransaction();

            TransactionalQueue<String> queue = context.getQueue("test");
            String queue_data = queue.poll();
            assertNotNull(queue_data);
            context.commitTransaction();
        }
    }

    static class MyQueueStore implements QueueStore, Serializable {

        static final Map<Long, Object> map = new HashMap<Long, Object>();

        static {
            map.put(1L, "hola");
            map.put(3L, "dias");
            map.put(4L, "pescado");
            map.put(6L, "oso");
            map.put(2L, "manzana");
            map.put(10L, "manana");
            map.put(12L, "perro");
            map.put(17L, "gato");
            map.put(19L, "toro");
            map.put(15L, "tortuga");
        }

        public void store(Long key, Object value) {
            map.put(key, value);
        }

        public void storeAll(Map map) {
            map.putAll(map);
        }

        public void delete(Long key) {
            map.remove(key);
        }

        public void deleteAll(Collection keys) {
            for (Object key : keys) {
                map.remove(key);
            }
        }

        public Object load(Long key) {
            return map.get(key);
        }

        public Map loadAll(Collection keys) {
            Map m = new HashMap();
            for (Object key : keys) {
                m.put(key, map.get(key));
            }
            return m;
        }

        public Set<Long> loadAllKeys() {
            return map.keySet();
        }
    }

    @Test
    public void testQueueStore() {
        Config config = new Config();
        int maxSize = 2000;
        QueueConfig queueConfig = config.getQueueConfig("testQueueStore");
        queueConfig.setMaxSize(maxSize);
        TestQueueStore queueStore = new TestQueueStore(1000, 0, 2000, 0, 0, 0, 1);

        QueueStoreConfig queueStoreConfig = new QueueStoreConfig();
        queueStoreConfig.setStoreImplementation(queueStore);
        queueConfig.setQueueStoreConfig(queueStoreConfig);

        TestHazelcastInstanceFactory factory = createHazelcastInstanceFactory(2);
        HazelcastInstance instance = factory.newHazelcastInstance(config);

        for (int i = 0; i < maxSize / 2; i++) {
            queueStore.store.put((long) i, i);
        }

        IQueue<Object> queue = instance.getQueue("testQueueStore");

        for (int i = 0; i < maxSize / 2; i++) {
            queue.offer(i + maxSize / 2);
        }

        instance.getLifecycleService().shutdown();
        HazelcastInstance instance2 = factory.newHazelcastInstance(config);

        IQueue<Object> queue2 = instance2.getQueue("testQueueStore");
        Assert.assertEquals(maxSize, queue2.size());


        Assert.assertEquals(maxSize, queueStore.store.size());
        for (int i = 0; i < maxSize; i++) {
            Assert.assertEquals(i, queue2.poll());
        }

        try {
            queueStore.assertAwait(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static class TestQueueStore implements QueueStore {

        final Map<Long, Integer> store = new LinkedHashMap<Long, Integer>();
        final CountDownLatch latchStore;
        final CountDownLatch latchStoreAll;
        final CountDownLatch latchDelete;
        final CountDownLatch latchDeleteAll;
        final CountDownLatch latchLoad;
        final CountDownLatch latchLoadAllKeys;
        final CountDownLatch latchLoadAll;
        final AtomicInteger callCount = new AtomicInteger();
        final AtomicInteger initCount = new AtomicInteger();
        final AtomicInteger destroyCount = new AtomicInteger();

        private boolean loadAllKeys = true;

        public TestQueueStore() {
            this(0, 0, 0, 0, 0, 0);
        }

        public TestQueueStore(int expectedStore, int expectedDelete, int expectedLoad) {
{'fraction_non_alphanumeric': 0.09690325678162759, 'fraction_numerical': 0.012402976714411458, 'mean_word_length': 2.973926868044515, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6439144', 'n_tokens_mistral': 3616, 'n_tokens_neox': 3434, 'n_words': 854}