content: package com.alibaba.otter.canal.store.memory.buffer;
  
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.TimeUnit;
  
  import org.junit.Assert;
  import org.junit.Test;
  
  import com.alibaba.otter.canal.protocol.position.Position;
  import com.alibaba.otter.canal.store.CanalStoreException;
  import com.alibaba.otter.canal.store.helper.CanalEventUtils;
  import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer;
  import com.alibaba.otter.canal.store.model.Event;
  import com.alibaba.otter.canal.store.model.Events;
  
  /**
   * 测试普通的put / get操作
   * 
   * @author jianghang 2012-6-19 下午09:50:08
   * @version 1.0.0
   */
  public class MemoryEventStorePutAndGetTest extends MemoryEventStoreBase {
  
      @Test
      public void testOnePut() {
          MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
          eventStore.start();
          // 尝试阻塞
          try {
              eventStore.put(buildEvent("1", 1L, 1L));
          } catch (Exception e) {
              Assert.fail(e.getMessage());
          }
          // 尝试阻塞+超时
          boolean result = false;
          try {
              result = eventStore.put(buildEvent("1", 1L, 1L), 1000L, TimeUnit.MILLISECONDS);
              Assert.assertTrue(result);
          } catch (Exception e) {
              Assert.fail(e.getMessage());
          }
          // 尝试
          result = eventStore.tryPut(buildEvent("1", 1L, 1L));
          Assert.assertTrue(result);
  
          eventStore.stop();
      }
  
      @Test
      public void testFullPut() {
          int bufferSize = 16;
          MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
          eventStore.setBufferSize(bufferSize);
          eventStore.start();
  
          for (int i = 0; i < bufferSize; i++) {
              boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
              Assert.assertTrue(result);
          }
  
          boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + bufferSize));
          Assert.assertFalse(result);
  
          try {
              result = eventStore.put(buildEvent("1", 1L, 1L + bufferSize), 1000L, TimeUnit.MILLISECONDS);
          } catch (CanalStoreException | InterruptedException e) {
              Assert.fail(e.getMessage());
          }
  
          Assert.assertFalse(result);
  
          eventStore.stop();
      }
  
      @Test
      public void testOnePutOneGet() {
          MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
          eventStore.start();
  
          boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L));
          Assert.assertTrue(result);
  
          Position position = eventStore.getFirstPosition();
          Events<Event> entrys = eventStore.tryGet(position, 1);
          Assert.assertTrue(entrys.getEvents().size() == 1);
          Assert.assertEquals(position, entrys.getPositionRange().getStart());
          Assert.assertEquals(position, entrys.getPositionRange().getEnd());
  
          eventStore.stop();
      }
  
      @Test
      public void testFullPutBatchGet() {
          int bufferSize = 16;
          MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
          eventStore.setBufferSize(bufferSize);
          eventStore.start();
  
          for (int i = 0; i < bufferSize; i++) {
              boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
              sleep(100L);
              Assert.assertTrue(result);
          }
  
          Position first = eventStore.getFirstPosition();
          Position lastest = eventStore.getLatestPosition();
          Assert.assertEquals(first, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L)));
          Assert.assertEquals(lastest, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L + bufferSize - 1)));
  
          System.out.println("start get");
          Events<Event> entrys1 = eventStore.tryGet(first, bufferSize);
          System.out.println("first get size : " + entrys1.getEvents().size());
  
          Assert.assertTrue(entrys1.getEvents().size() == bufferSize);
          Assert.assertEquals(first, entrys1.getPositionRange().getStart());
          Assert.assertEquals(lastest, entrys1.getPositionRange().getEnd());
  
          Assert.assertEquals(first, CanalEventUtils.createPosition(entrys1.getEvents().get(0)));
          Assert.assertEquals(lastest, CanalEventUtils.createPosition(entrys1.getEvents().get(bufferSize - 1)));
          eventStore.stop();
      }
  
      @Test
      public void testBlockPutOneGet() {
          final MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
          eventStore.start();
  
          final int batchSize = 10;
          for (int i = 0; i < batchSize; i++) {
              boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L));
              Assert.assertTrue(result);
          }
  
          final Position position = eventStore.getFirstPosition();
          try {
              Events<Event> entrys = eventStore.get(position, batchSize);
              Assert.assertTrue(entrys.getEvents().size() == batchSize);
              Assert.assertEquals(position, entrys.getPositionRange().getStart());
              Assert.assertEquals(position, entrys.getPositionRange().getEnd());
          } catch (CanalStoreException | InterruptedException e) {
          }
  
          ExecutorService executor = Executors.newFixedThreadPool(1);
          executor.submit(() -> {
              boolean result = false;
              try {
                  eventStore.get(position, batchSize);
              } catch (CanalStoreException e) {
              } catch (InterruptedException e) {
                  System.out.println("interrupt occured.");
                  result = true;
              }
              Assert.assertTrue(result);
          });
  
          try {
              Thread.sleep(1000L);
          } catch (InterruptedException e) {
              Assert.fail(e.getMessage());
          }
          executor.shutdownNow();
  
          try {
              Thread.sleep(1000L);
          } catch (InterruptedException e) {
              Assert.fail(e.getMessage());
          }
          eventStore.stop();
      }
  }
  