/*

* 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 javaobject;

import java.util.*;

import org.apache.geode.cache.CacheWriter;

import org.apache.geode.cache.EntryEvent;

import org.apache.geode.cache.Operation;

import org.apache.geode.cache.RegionEvent;

import org.apache.geode.cache.CacheFactory;

import org.apache.geode.cache.Declarable;

import org.apache.geode.internal.cache.tier.sockets.CacheClientProxy;

import org.apache.geode.LogWriter;

import org.apache.geode.cache.CacheWriterException;

import org.apache.geode.internal.cache.PartitionedRegion;

import org.apache.geode.cache.partition.PartitionRegionHelper;

import org.apache.geode.internal.cache.PartitionedRegionHelper;

import org.apache.geode.cache.Cache;

import org.apache.geode.cache.CacheException;

import org.apache.geode.internal.cache.EntryEventImpl;

import java.util.List;

public class CacheWriterForSingleHop<K,V> implements Declarable, CacheWriter<K,V> {

public CacheWriterForSingleHop()

{

}

public void init(Properties props) {

}

public void checkSingleHop(EntryEvent<K,V> event) {

checkSingleHop(event, true);

}

public void checkSingleHop(EntryEvent<K,V> event, boolean checkhash) {

System.out.println("CPPTEST: java key: " + event.getKey());

System.out.println("CPPTEST: java newvalue: " + event.getNewValue());

System.out.println("CPPTEST: java hashcode: " + event.getKey().hashCode());

System.out.println("CPPTEST: operation: " + event.getOperation());

if ((event.getNewValue() instanceof Boolean) || (event.getNewValue() instanceof String)) {

System.out.println("CPPTEST: Short-circuiting CacheWriter because its a special put.");

return;

}

/*

if (event.getOperation() == Operation.LOCAL_LOAD_CREATE) {

System.out.println("CPPTEST: Detected NWHOP for LOCAL_LOAD_CREATE.");

return;

}

*/

if (event.isOriginRemote()) {

System.out.println("CPPTEST: Detected NWHOP via isOriginRemote() for operation " + event.getOperation());

return;

}

if (checkhash) {

int hashcode = event.getKey().hashCode();

int value = Integer.parseInt( event.getNewValue().toString() );

if (event.getKey().hashCode() != value){

System.out.println("CPPTEST: hashcode did not match value");

throw new CacheWriterException("java value " + event.getNewValue().toString() +

" does not match " + event.getKey().hashCode() + " java hashcode" );

//return;

}

System.out.println("CPPTEST: hashcode check OK");

}

PartitionedRegion pr;

try {

pr = (PartitionedRegion) ((EntryEventImpl)event).getRegion();

} catch (Exception ex) {

throw new CacheWriterException("PartitionRegion Cast failed due to: "+ ex.getMessage());

//System.out.println("CPPTEST: PartitionRegion Cast failed due to: "+ ex.getMessage());

//return;

}

Integer bucketId = new Integer(PartitionedRegionHelper.getHashKey(pr, null, event.getKey(), null, null));

System.out.println("CPPTEST: found BucketId: " + bucketId);

List bucketListOnNode = pr.getLocalPrimaryBucketsListTestOnly();

System.out.println("CPPTEST: bucketListOnNode size is: " + bucketListOnNode.size());

/*

if (!bucketListOnNode.contains(bucketId)) {

System.out.println("CPPTEST: Wrong BucketId from list");

//throw new CacheWriterException("Calculated bucketId " + bucketId +

//" does not match java partitionedRegion BucketId" );

return;

}

*/

if (PartitionRegionHelper.getPrimaryMemberForKey(pr, event.getKey()) == null) {

System.out.println("CPPTEST: Wrong BucketId from member");

throw new CacheWriterException("Calculated bucketId " + bucketId +

" does not match java partitionedRegion BucketId" );

//return;

}

System.out.println("CPPTEST: cache writer OK BucketId");

// We need to put this into the region here because throwing an exception

// aborts the put thus causing the subsequent destroy op to fail because

// the CacheWriter itself is not invoked due to EntryNotFoundException.

//pr.putIfAbsent(event.getKey(), "placeholder");

/*

We operate backwards, throwing exception for success conditions

and simply returning for failure conditions due to the way

PR do not (upon extra hop) invoke writers/loaders on secondaries.

*/

/*throw new CacheWriterException("Calculated bucketId " + bucketId +

" matches java partitionedRegion primary BucketId" );*/

}

public void beforeCreate(EntryEvent<K,V> event) {

System.out.println("CPPTEST: Inside beforeCreate");

checkSingleHop(event);

}

public void beforeUpdate(EntryEvent<K,V> event)

{

System.out.println("CPPTEST: Inside beforeUpdate");

checkSingleHop(event);

}

public void beforeDestroy(EntryEvent<K,V> event)

{

System.out.println("CPPTEST: Inside beforeDestroy");

checkSingleHop(event, false);

}

public void beforeRegionDestroy(RegionEvent<K,V> event)

{

}

public void beforeRegionClear(RegionEvent<K,V> event)

{

}

public void close()

{

}

}