package com.hazelcast.client.multimap;

import com.hazelcast.client.HazelcastClient;
import com.hazelcast.core.*;
import com.hazelcast.test.HazelcastParallelClassRunner;
import com.hazelcast.test.annotation.NightlyTest;
import com.hazelcast.test.annotation.ProblematicTest;
import com.hazelcast.test.annotation.QuickTest;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;

import java.util.concurrent.atomic.AtomicInteger;

import static com.hazelcast.test.HazelcastTestSupport.*;
import static com.hazelcast.test.HazelcastTestSupport.randomString;
import static junit.framework.Assert.assertEquals;

@RunWith(HazelcastParallelClassRunner.class)
@Category(QuickTest.class)
public class ClientMultiMapListenerStressTest {

    static final String MAP_NAME = randomString();
    static final int NUMBER_OF_CLIENTS = 1;
    static final int THREADS_PER_CLIENT = 8;

    static HazelcastInstance server;

    @BeforeClass
    public static void init() {
        server = Hazelcast.newHazelcastInstance();
    }

    @AfterClass
    public static void destroy() {
        HazelcastClient.shutdownAll();
        Hazelcast.shutdownAll();
    }


    @Test
    @Category(ProblematicTest.class)
    public void listenerAddStressTest() throws InterruptedException {

        PutItemsThread[] putThreads = new PutItemsThread[NUMBER_OF_CLIENTS * THREADS_PER_CLIENT];

        int idx=0;
        for(int i=0; i<NUMBER_OF_CLIENTS; i++){
            HazelcastInstance client = HazelcastClient.newHazelcastClient();
            for(int j=0; j<THREADS_PER_CLIENT; j++){
                PutItemsThread t = new PutItemsThread(client);
                putThreads[idx++]=t;
            }
        }

        for(int i=0; i<putThreads.length; i++){
            putThreads[i].start();
        }

        assertJoinable(putThreads);

        MultiMap mm = server.getMultiMap(MAP_NAME);
        assertEquals(PutItemsThread.maxItems * putThreads.length, mm.size());

        for(int i=0; i<putThreads.length; i++){
            putThreads[i].assertResult(PutItemsThread.maxItems * putThreads.length);
        }
    }

    public class PutItemsThread extends Thread{

        public static final int maxItems = 1000;

        public final MyEntryListener listener = new MyEntryListener();
        public HazelcastInstance hzInstance;
        public MultiMap mm;
        public String id;

        public PutItemsThread(HazelcastInstance hzInstance){
            this.id = randomString();
            this.hzInstance = hzInstance;
            this.mm = hzInstance.getMultiMap(MAP_NAME);
            mm.addEntryListener(listener, true);
        }

        public void run(){
            for(int i=0; i<maxItems; i++){
                mm.put(id+i, id+i);
            }
        }

        public void assertResult(int target){
            assertEquals(target, listener.add.get());
        }
    }

    static class MyEntryListener implements EntryListener {

        public AtomicInteger add = new AtomicInteger(0);

        public void entryAdded(EntryEvent event) {
            add.incrementAndGet();
        }

        public void entryRemoved(EntryEvent event) {
        }

        public void entryUpdated(EntryEvent event) {

        }

        public void entryEvicted(EntryEvent event) {

        }
    };
}
@@ -103,24 +103,62 @@
 ast.core.*;%0A
+import com.hazelcast.test.AssertTask;%0A
 import com.h
@@ -255,62 +255,8 @@
 st;%0A
-import com.hazelcast.test.annotation.ProblematicTest;%0A
 impo
@@ -749,21 +749,23 @@
 ategory(
-Quick
+Nightly
 Test.cla
@@ -813,24 +813,68 @@
 ressTest %7B%0A%0A
+    static final int MAX_SECONDS = 60 * 10;%0A
     static f
@@ -953,17 +953,17 @@
 IENTS = 
-1
+8
 ;%0A    st
@@ -998,17 +998,18 @@
 LIENT = 
-8
+16
 ;%0A%0A    s
@@ -1293,45 +1293,8 @@
 est%0A
-    @Category(ProblematicTest.class)%0A
     
@@ -1360,32 +1360,38 @@
 tion %7B%0A%0A        
+final 
 PutItemsThread%5B%5D
@@ -1904,16 +1904,29 @@
 oinable(
+MAX_SECONDS, 
 putThrea
@@ -1927,16 +1927,17 @@
 tThreads
+ 
 );%0A%0A    
@@ -2016,32 +2016,33 @@
 ItemsThread.
-maxItems
+MAX_ITEMS
  * putThread
@@ -2060,24 +2060,130 @@
 m.size());%0A%0A
+        assertTrueEventually(new AssertTask() %7B%0A%0A            public void run() throws Exception %7B%0A        
         for(
@@ -2222,32 +2222,40 @@
 +)%7B%0A            
+   
+     
 putThreads%5Bi%5D.as
@@ -2280,24 +2280,25 @@
 sThread.
-maxItems
+MAX_ITEMS
  * putTh
@@ -2312,33 +2312,68 @@
 ength);%0A        
-%7D
+        %7D%0A            %7D%0A        %7D);%0A
 %0A    %7D%0A%0A    publ
@@ -2445,24 +2445,25 @@
 nal int 
-maxItems
+MAX_ITEMS
  = 1000;
@@ -2950,16 +2950,18 @@
 ; i%3C
-maxItems
+ MAX_ITEMS
 ; i+

fixed with assert true Eventual