/*
 * Copyright © 2015 Cask Data, Inc.
 *
 * 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 co.cask.cdap.data.view;

import co.cask.cdap.common.NotFoundException;
import co.cask.cdap.proto.Id;
import co.cask.cdap.proto.ViewDetail;
import co.cask.cdap.proto.ViewSpecification;
import com.google.common.collect.HashBasedTable;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Table;

import java.util.List;

/**
 * In-memory implementation of {@link ViewStore}.
 */
public final class InMemoryViewStore implements ViewStore {

  private final Table<Id.Stream.View, Id.Stream, ViewSpecification> views;

  public InMemoryViewStore() {
    views = HashBasedTable.create();
  }

  @Override
  public boolean createOrUpdate(Id.Stream.View viewId, ViewSpecification config) {
    boolean created = !views.containsRow(viewId);
    views.put(viewId, viewId.getStream(), config);
    return created;
  }

  @Override
  public boolean exists(Id.Stream.View viewId) {
    return views.contains(viewId, viewId.getStream());
  }

  @Override
  public void delete(Id.Stream.View viewId) throws NotFoundException {
    ViewSpecification removed = views.remove(viewId, viewId.getStream());
    if (removed == null) {
      throw new NotFoundException(viewId);
    }
  }

  @Override
  public List<Id.Stream.View> list(Id.Stream streamId) {
    return ImmutableList.<Id.Stream.View>builder().addAll(views.column(streamId).keySet()).build();
  }

  @Override
  public ViewDetail get(Id.Stream.View viewId) throws NotFoundException {
    if (!views.containsRow(viewId)) {
      throw new NotFoundException(viewId);
    }

    return new ViewDetail(viewId.getId(), views.get(viewId, viewId.getStream()));
  }
}

@@ -943,16 +943,163 @@
 il.List;
+%0Aimport java.util.concurrent.locks.Lock;%0Aimport java.util.concurrent.locks.ReadWriteLock;%0Aimport java.util.concurrent.locks.ReentrantReadWriteLock;
 %0A%0A/**%0A *
@@ -1285,16 +1285,57 @@
 %3E views;
+%0A  private final ReadWriteLock viewsLock;
 %0A%0A  publ
@@ -1359,24 +1359,29 @@
 ore() %7B%0A    
+this.
 views = Hash
@@ -1397,24 +1397,75 @@
 e.create();%0A
+    this.viewsLock = new ReentrantReadWriteLock();%0A
   %7D%0A%0A  @Over
@@ -1560,124 +1560,180 @@
     
-boolean created
+Lock lock
  = 
-!
 views
-.containsRow(viewId);%0A    views.put(viewId, viewId.getStream(), config);%0A    return created;
+Lock.writeLock();%0A    lock.lock();%0A    try %7B%0A      return views.put(viewId, viewId.getStream(), config) == null;%0A    %7D finally %7B%0A      lock.unlock();%0A    %7D
 %0A  %7D
@@ -1791,24 +1791,91 @@
 w viewId) %7B%0A
+    Lock lock = viewsLock.readLock();%0A    lock.lock();%0A    try %7B%0A  
     return v
@@ -1913,24 +1913,67 @@
 tStream());%0A
+    %7D finally %7B%0A      lock.unlock();%0A    %7D%0A
   %7D%0A%0A  @Over
@@ -2070,16 +2070,97 @@
 ication 
+removed;%0A    Lock lock = viewsLock.writeLock();%0A    lock.lock();%0A    try %7B%0A      
 removed 
@@ -2199,24 +2199,67 @@
 tStream());%0A
+    %7D finally %7B%0A      lock.unlock();%0A    %7D%0A
     if (remo
@@ -2392,24 +2392,91 @@
 streamId) %7B%0A
+    Lock lock = viewsLock.readLock();%0A    lock.lock();%0A    try %7B%0A  
     return I
@@ -2559,24 +2559,67 @@
 )).build();%0A
+    %7D finally %7B%0A      lock.unlock();%0A    %7D%0A
   %7D%0A%0A  @Over
@@ -2693,24 +2693,91 @@
 Exception %7B%0A
+    Lock lock = viewsLock.readLock();%0A    lock.lock();%0A    try %7B%0A  
     if (!vie
@@ -2798,24 +2798,26 @@
 (viewId)) %7B%0A
+  
       throw 
@@ -2851,19 +2851,23 @@
 d);%0A    
+  
 %7D%0A%0A
+  
     retu
@@ -2940,14 +2940,57 @@
 am()));%0A
+    %7D finally %7B%0A      lock.unlock();%0A    %7D%0A
   %7D%0A%7D%0A

Make InMemoryViewStore thread safe