/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.hedwig.server.persistence;

import java.util.HashSet;
import java.util.Set;

import org.apache.hedwig.protocol.PubSubProtocol.Message;

/**
 * This class is NOT thread safe. It need not be thread-safe because our
 * read-ahead cache will operate with only 1 thread
 *
 */
public class CacheValue {

    // Actually we don't care the order of callbacks
    // when a scan callback, it should be delivered to both callbacks
    Set<ScanCallbackWithContext> callbacks;
    Message message;

    public CacheValue() {
        this.callbacks = new HashSet<ScanCallbackWithContext>();
    }

    public CacheValue(Message message) {
        this.message = message;
    }

    synchronized public boolean isStub() {
        return message == null;
    }

    synchronized public boolean wasStub() {
        return callbacks != null;
    }

    // Cache weight static (loading cache)
    synchronized public int getCacheWeight() {
        return wasStub()? 0 : message.getBody().size();
    }

    synchronized public void setMessageAndInvokeCallbacks(Message message) {
        if (this.message != null) {
            return;
        }

        this.message = message;

        for (ScanCallbackWithContext callbackWithCtx : callbacks) {
            if (null != callbackWithCtx) {
                callbackWithCtx.getScanCallback().messageScanned(callbackWithCtx.getCtx(), message);
            }
        }
    }

    synchronized public boolean removeCallback(ScanCallback callback, Object ctx) {
        if (null == callbacks) {
            return false;
        }
        return callbacks.remove(new ScanCallbackWithContext(callback, ctx));
    }

    synchronized public void addCallback(ScanCallback callback, Object ctx) {
        if (!isStub()) {
            // call the callback right away
            callback.messageScanned(ctx, message);
            return;
        }

        callbacks.add(new ScanCallbackWithContext(callback, ctx));
    }

    synchronized public Message getMessage() {
        return message;
    }

    synchronized public void setErrorAndInvokeCallbacks(Exception exception) {
        if (this.message != null) {
            return;
        }

        for (ScanCallbackWithContext callbackWithCtx : callbacks) {
            if (null != callbackWithCtx) {
                callbackWithCtx.getScanCallback().scanFailed(callbackWithCtx.getCtx(), exception);
            }
        }
    }
}

@@ -847,16 +847,46 @@
 tence;%0A%0A
+import java.util.Collections;%0A
 import j
@@ -1388,16 +1388,44 @@
 backs = 
+Collections.synchronizedSet(
 new Hash
@@ -1454,16 +1454,17 @@
 ntext%3E()
+)
 ;%0A    %7D%0A
@@ -1544,37 +1544,24 @@
 %0A    %7D%0A%0A    
-synchronized 
 public boole
@@ -1613,37 +1613,24 @@
 %0A    %7D%0A%0A    
-synchronized 
 public boole
@@ -1728,37 +1728,24 @@
  cache)%0A    
-synchronized 
 public int g
@@ -1825,37 +1825,24 @@
 %0A    %7D%0A%0A    
-synchronized 
 public void 
@@ -1984,25 +1984,63 @@
  = message;%0A
-%0A
+        synchronized (callbacks) %7B%0A    
         for 
@@ -2099,32 +2099,36 @@
 ) %7B%0A            
+ 
+   
 if (null != call
@@ -2134,32 +2134,36 @@
 lbackWithCtx) %7B%0A
+    
                 
@@ -2239,32 +2239,50 @@
 tx(), message);%0A
+                %7D%0A
             %7D%0A  
@@ -2296,37 +2296,24 @@
 %0A    %7D%0A%0A    
-synchronized 
 public boole
@@ -2520,37 +2520,24 @@
 %0A    %7D%0A%0A    
-synchronized 
 public void 
@@ -2810,37 +2810,24 @@
 %0A    %7D%0A%0A    
-synchronized 
 public Messa
@@ -2879,29 +2879,16 @@
  %7D%0A%0A    
-synchronized 
 public v

Undo sync in CacheValue - Undo sync in CacheValue