/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 *  contributor license agreements.  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.  For additional information regarding
 * copyright in this work, please see the NOTICE file in the top level
 * directory of this distribution.
 */

package org.apache.usergrid.persistence.collection.mvcc.entity.impl;

import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import org.apache.usergrid.persistence.collection.guice.MvccEntityDelete;
import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
import org.apache.usergrid.persistence.core.consistency.AsyncProcessor;
import org.apache.usergrid.persistence.core.consistency.MessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Observable;
import rx.functions.Func1;

/**
 * Listens for delete entity event then deletes entity for real this time
 */
public class MvccEntityDeleteListener implements MessageListener<MvccEntityEvent<MvccEntity>,MvccEntityEvent<MvccEntity>> {

    private static final Logger LOG = LoggerFactory.getLogger(MvccEntityDeleteListener.class);

    private final MvccEntitySerializationStrategy entityMetadataSerialization;

    public MvccEntityDeleteListener(final MvccEntitySerializationStrategy entityMetadataSerialization,
                                    @MvccEntityDelete final AsyncProcessor<MvccEntityEvent<MvccEntity>> entityDelete){
        this.entityMetadataSerialization = entityMetadataSerialization;
        entityDelete.addListener( this );
    }

    @Override
    public Observable<MvccEntityEvent<MvccEntity>> receive(final MvccEntityEvent<MvccEntity> entityEvent) {
        final MvccEntity entity = entityEvent.getData();
        return Observable.from(entity).map( new Func1<MvccEntity, MvccEntityEvent<MvccEntity>>() {
            @Override
            public MvccEntityEvent<MvccEntity> call(MvccEntity mvccEntity) {
                try {
                    entityMetadataSerialization.delete(entityEvent.getCollectionScope(),entity.getId(),entity.getVersion()).execute();
                    LOG.info("Successfully deleted entity",entity);
                }catch(ConnectionException e){
                    LOG.error("Unable to delete entity from listener",entity);
                    throw new RuntimeException( "Unable to delete entity - " + entity, e );
                }
                return entityEvent;
            }
        });
    }
}

@@ -2522,34 +2522,20 @@
 y, M
-vccEntityEvent%3CMvccEntity%3E
+utationBatch
 %3E() 
@@ -2582,34 +2582,20 @@
 ic M
-vccEntityEvent%3CMvccEntity%3E
+utationBatch
  cal
@@ -2640,34 +2640,14 @@
     
- try %7B%0A                   
+return
  ent
@@ -2750,18 +2750,8 @@
 n())
-.execute()
 ;%0A  
@@ -2756,32 +2756,34 @@
             
+%7D%0A
         
 LOG.info(%22Su
@@ -2774,181 +2774,373 @@
     
-LOG.info(%22Successfully deleted entity%22,entity);%0A                %7Dcatch(ConnectionException e)%7B%0A                    LOG.error(%22Unable to delete entity from listener%22,entity);
+%7D).map(new Func1%3CMutationBatch, MvccEntityEvent%3CMvccEntity%3E%3E() %7B%0A            @Override%0A            public MvccEntityEvent%3CMvccEntity%3E call(final MutationBatch mutationBatch) %7B%0A%0A                //actually delete the edge from both the commit log and%0A                try %7B%0A                    mutationBatch.execute();%0A                %7D catch (ConnectionException e) %7B
 %0A   
@@ -3183,17 +3183,16 @@
 ception(
- 
 %22Unable 
@@ -3198,38 +3198,28 @@
  to 
-delete entity - %22 + entity
+execute mutation%22
 , e
- 
 );%0A 
@@ -3227,32 +3227,33 @@
               %7D%0A
+%0A
                 

change test