/*
 * Copyright 2000-2017 JetBrains s.r.o.
 *
 * 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.intellij.openapi.vcs.changes.committed;

import com.intellij.openapi.application.ApplicationManager;
import com.intellij.openapi.util.Ref;
import com.intellij.testFramework.UsefulTestCase;
import com.intellij.testFramework.vcs.FileBasedTest;
import com.intellij.util.BufferedListConsumer;
import com.intellij.util.ConcurrencyUtil;
import com.intellij.util.concurrency.Semaphore;
import org.jetbrains.annotations.NotNull;
import org.junit.Test;

import java.util.*;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class BufferedListConsumerTest extends FileBasedTest {

  @Test
  public void testHugeWriteRead() throws Exception {
    Ref<Future<?>> futureRef = new Ref<>();

    final Random random = new Random(17);
    final Set<Long> src = new HashSet<>(200);
    for (int i = 0; i < 100; i++) {
      src.add(System.currentTimeMillis());
      src.add(random.nextLong());
    }
    final List<Long> dst = new ArrayList<>();
    final BufferedListConsumer<Long> consumer = new BufferedListConsumer<Long>(9, items -> dst.addAll(items), 4) {
      @Override
      protected void invokeConsumer(@NotNull Runnable consumerRunnable) {
        futureRef.set(ApplicationManager.getApplication().executeOnPooledThread(consumerRunnable));
      }
    };
    final Semaphore semaphore = new Semaphore();
    semaphore.down();
    Thread thread = new Thread("buffered list test") {
      @Override
      public void run() {
        for (Long aLong : src) {
          consumer.consumeOne(aLong);
        }
        consumer.flush();
        semaphore.up();
      }
    };
    thread.start();

    int timeout = 10 * 1000;
    try {
      if (!semaphore.waitFor(timeout)) throw new Exception("Couldn't await background thread");

      Future<?> future = futureRef.get();
      if (future != null) future.get(timeout, TimeUnit.MILLISECONDS);

      UsefulTestCase.assertOrderedEquals(dst, src);
    }
    finally {
      ConcurrencyUtil.joinAll(thread);
    }
  }
}

@@ -1,13 +1,10 @@
 /
-*%0A *
+/
  Copyrig
@@ -14,17 +14,17 @@
 2000-201
-7
+8
  JetBrai
@@ -36,767 +36,162 @@
 r.o.
-%0A *%0A * Licensed under the Apache License, Version 2.0 (the %22License%22);%0A * you may not use this file except in compliance with the License.%0A * You may obtain a copy of the License at%0A *%0A * http://www.apache.org/licenses/LICENSE-2.0%0A *%0A * Unless required by applicable law or agreed to in writing, software%0A * distributed under the License is distributed on an %22AS IS%22 BASIS,%0A * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A * See the License for the specific language governing permissions and%0A * limitations under the License.%0A */%0Apackage com.intellij.openapi.vcs.changes.committed;%0A%0Aimport com.intellij.openapi.application.ApplicationManager;%0Aimport com.intellij.openapi.util.Ref;%0Aimport com.intellij.testFramework.UsefulTestCase;
+ Use of this source code is governed by the Apache 2.0 license that can be found in the LICENSE file.%0Apackage com.intellij.openapi.vcs.changes.committed;%0A
 %0Aimp
@@ -315,28 +315,21 @@
 util.Con
-currencyUtil
+sumer
 ;%0Aimport
@@ -374,50 +374,8 @@
 re;%0A
-import org.jetbrains.annotations.NotNull;%0A
 impo
@@ -418,34 +418,126 @@
 .*;%0A
+%0A
 import 
-java
+static com.intellij.testFramework.UsefulTestCase.assertSameElements;%0Aimport static com.intellij
 .util.
-c
+C
 oncurren
 t.Fu
@@ -536,16 +536,22 @@
 rren
-t.Future
+cyUtil.joinAll
 ;%0Aim
@@ -547,32 +547,39 @@
 joinAll;%0Aimport 
+static 
 java.util.concur
@@ -576,26 +576,35 @@
 til.
-concurrent.TimeUni
+Collections.synchronizedLis
 t;%0A%0A
@@ -731,53 +731,8 @@
 n %7B%0A
-    Ref%3CFuture%3C?%3E%3E futureRef = new Ref%3C%3E();%0A%0A
     
@@ -948,25 +948,81 @@
 nal 
-List%3CLong%3E dst = 
+int sourceSize = src.size();%0A    final List%3CLong%3E dst = synchronizedList(
 new 
@@ -1034,16 +1034,17 @@
 List%3C%3E()
+)
 ;%0A    fi
@@ -1051,79 +1051,87 @@
 nal 
-BufferedListConsumer%3CLong%3E consumer = new BufferedListConsumer%3CLong%3E(9,
+Semaphore semaphore = new Semaphore();%0A    Consumer%3CList%3CLong%3E%3E innerConsumer =
  ite
@@ -1135,16 +1135,24 @@
 items -%3E
+ %7B%0A     
  dst.add
@@ -1165,288 +1165,181 @@
 ems)
-, 4) %7B
+;
 %0A      
-@Override%0A      protected void invokeConsumer(@NotNull Runnable consumerRunnable) %7B%0A        futureRef.set(ApplicationManager.getApplication().executeOnPooledThread(consumerRunnable));%0A      %7D%0A    %7D;%0A    final Semaphore semaphore = new Semaphore();%0A    semaphore.down(
+if (sourceSize == dst.size()) %7B%0A        semaphore.up();%0A      %7D%0A    %7D;%0A    final BufferedListConsumer%3CLong%3E consumer = new BufferedListConsumer%3C%3E(9, innerConsumer, 4
 );%0A 
@@ -1541,24 +1541,35 @@
 lush();%0A    
+  %7D%0A    %7D;%0A
     semaphor
@@ -1570,35 +1570,22 @@
 maphore.
-up();%0A      %7D%0A    %7D
+down()
 ;%0A    th
@@ -1745,154 +1745,222 @@
     
-Future%3C?%3E future = futureRef.get();%0A      if (future != null) future.get(timeout, TimeUnit.MILLISECONDS);%0A%0A      UsefulTestCase.assertOrderedEqual
+// BufferedListConsumer calls inner consumer in some background thread. This thread could be not the same between calls, which could%0A      // lead to different elements order in dst and src.%0A      assertSameElement
 s(ds
@@ -1994,32 +1994,16 @@
 %7B%0A      
-ConcurrencyUtil.
 joinAll(

Fix synchronization and wait logic in "BufferedListConsumerTest"