/*
 * Copyright (c) 2008-2014, 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.map.mapstore.writebehind;

import com.hazelcast.core.IMap;
import com.hazelcast.core.MapStoreAdapter;
import com.hazelcast.test.AssertTask;
import com.hazelcast.test.HazelcastSerialClassRunner;
import com.hazelcast.test.HazelcastTestSupport;
import com.hazelcast.test.annotation.QuickTest;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;

import static org.junit.Assert.assertEquals;

@RunWith(HazelcastSerialClassRunner.class)
@Category(QuickTest.class)
public class WriteBehindFailAndRetryTest extends HazelcastTestSupport {

    @Test
    public void testStoreOperationDone_afterTemporaryMapStoreFailure() throws Exception {
        final SelfHealingMapStore mapStore = new SelfHealingMapStore<Integer, Integer>();
        final IMap map = TestMapUsingMapStoreBuilder.create()
                .withMapStore(mapStore)
                .withNodeCount(1)
                .withNodeFactory(createHazelcastInstanceFactory(1))
                .withWriteDelaySeconds(1)
                .withPartitionCount(1)
                .build();

        map.put(1, 2);

        assertTrueEventually(new AssertTask() {
            @Override
            public void run() throws Exception {
                assertEquals(1, mapStore.size());
            }
        });
    }


    static class SelfHealingMapStore<K, V> extends MapStoreAdapter<K, V> {

        private static final long DIFF = TimeUnit.SECONDS.toMillis(3);

        private ConcurrentMap<K, V> store = new ConcurrentHashMap<K, V>();

        private ConcurrentMap<String, Long> timeHolder = new ConcurrentHashMap<String, Long>();

        private static final String TIME_KEY = "time";

        @Override
        public void store(K key, V value) {
            final Object time = timeHolder.get(TIME_KEY);
            if (time == null) {
                timeHolder.put(TIME_KEY, System.currentTimeMillis());
            } else {
                final long now = System.currentTimeMillis();
                final long diff = now - (Long) time;
                if (diff > DIFF) {
                    store.put(key, value);
                    return;
                }
            }
            throw new RuntimeException("Temporary map store failure");
        }

        public int size() {
            return store.size();
        }
    }
}

@@ -928,24 +928,57 @@
 .QuickTest;%0A
+import com.hazelcast.util.Clock;%0A
 import org.j
@@ -2216,79 +2216,13 @@
 ate 
-static final long DIFF = TimeUnit.SECONDS.toMillis(3);%0A%0A        private
+final
  Con
@@ -2298,296 +2298,771 @@
 ate 
-ConcurrentMap%3CString, Long%3E timeHolder = new ConcurrentHashMap%3CString, Long%3E();%0A%0A        private static final String TIME_KEY = %22time%22;%0A%0A        @Override%0A        public void store(K key, V value) %7B%0A            final Object time = timeHolder.get(TIME_KEY);%0A            if (time == null) %7B
+final TemporarySuccessProducer temporarySuccessProducer%0A                = new TemporarySuccessProducer(10);%0A%0A        @Override%0A        public void store(K key, V value) %7B%0A            temporarySuccessProducer.successOrException();%0A%0A            store.put(key, value);%0A        %7D%0A%0A        public int size() %7B%0A            return store.size();%0A        %7D%0A    %7D%0A%0A%0A    @Test%0A    public void testName() throws Exception %7B%0A        TemporarySuccessProducer temporarySuccessProducer%0A                = new TemporarySuccessProducer(10);%0A        for (int i = 0; i %3C 1000; i++) %7B%0A            Thread.sleep(1000);%0A            try %7B%0A                temporarySuccessProducer.successOrException();%0A            %7D catch (Exception e) %7B%0A                e.printStackTrace();%0A            %7D%0A
 %0A   
@@ -3070,95 +3070,533 @@
     
+%7D%0A
     
+%7D%0A%0A
     
-timeHolder.put(TIME_KEY, System.currentTimeMillis());%0A            %7D else %7B%0A    
+/**%0A     * Produces periodic success, otherwise there will be many exceptions.%0A     */%0A    private static class TemporarySuccessProducer %7B%0A%0A        private final long successGenerationPeriodInMillis;%0A%0A        private volatile long startMillis;%0A%0A        public TemporarySuccessProducer(long secondsPeriod) %7B%0A            this.successGenerationPeriodInMillis = TimeUnit.SECONDS.toMillis(secondsPeriod);%0A            this.startMillis = Clock.currentTimeMillis();%0A        %7D%0A%0A        public void successOrException() %7B%0A
     
@@ -3624,14 +3624,13 @@
 w = 
-System
+Clock
 .cur
@@ -3651,36 +3651,32 @@
 ();%0A            
-    
 final long diff 
@@ -3674,116 +3674,164 @@
 ong 
-diff = now - (Long) time;%0A                if (diff %3E DIFF) %7B%0A                    store.put(key, value);%0A    
+elapsedTime = now - startMillis;%0A%0A            if (elapsedTime %3E successGenerationPeriodInMillis) %7B%0A                startMillis = Clock.currentTimeMillis();%0A
     
@@ -3866,27 +3866,10 @@
     
-    %7D%0A            
 %7D
+%0A
 %0A   
@@ -3891,126 +3891,215 @@
 new 
-RuntimeException(%22Temporary map store failure%22);%0A        %7D%0A%0A        public int size() %7B%0A            return store.size(
+TemporaryMapStoreException();%0A        %7D%0A    %7D%0A%0A    private static class TemporaryMapStoreException extends RuntimeException %7B%0A%0A        public TemporaryMapStoreException() %7B%0A            super(%22Test exception%22
 );%0A 
@@ -4113,10 +4113,12 @@
 %7D%0A    %7D%0A
+%0A%0A
 %7D%0A

test improvement