/*
 * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
 * and is available at http://www.eclipse.org/legal/epl-v10.html
 */

package org.opendaylight.controller.cluster.notifications;

import akka.actor.Actor;
import akka.actor.ActorPath;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.japi.Creator;
import akka.serialization.Serialization;
import com.google.common.collect.Maps;
import java.util.Map;
import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActor;

/**
 * The RoleChangeNotifier is responsible for receiving Raft role change messages and notifying
 * the listeners (within the same node), which are registered with it.
 * <p/>
 * The RoleChangeNotifier is instantiated by the Shard and injected into the RaftActor.
 */
public class RoleChangeNotifier extends AbstractUntypedActor implements AutoCloseable {

    private String memberId;
    private Map<ActorPath, ActorRef> registeredListeners = Maps.newHashMap();
    private RoleChangeNotification latestRoleChangeNotification = null;

    public RoleChangeNotifier(String memberId) {
        this.memberId = memberId;
    }

    public static Props getProps(final String memberId) {
        return Props.create(new Creator<Actor>() {
            @Override
            public Actor create() throws Exception {
                return new RoleChangeNotifier(memberId);
            }
        });
    }

    @Override
    public void preStart() throws Exception {
        super.preStart();
        LOG.info("RoleChangeNotifier:{} created and ready for shard:{}",
            Serialization.serializedActorPath(getSelf()), memberId);
    }

    @Override
    protected void handleReceive(Object message) throws Exception {
        if (message instanceof RegisterRoleChangeListener) {
            // register listeners for this shard

            ActorRef curRef = registeredListeners.get(getSender().path());
            if (curRef != null) {
                // ActorPaths would pass equal even if the unique id of the actors are different
                // if a listener actor is re-registering after reincarnation, then removing the existing
                // entry so the actor path with correct unique id is registered.
                registeredListeners.remove(getSender().path());
            }
            registeredListeners.put(getSender().path(), getSender());

            LOG.info("RoleChangeNotifier for {} , registered listener {}", memberId,
                getSender().path().toString());

            getSender().tell(new RegisterRoleChangeListenerReply(), getSelf());

            if (latestRoleChangeNotification != null) {
                getSender().tell(latestRoleChangeNotification, getSelf());
            }


        } else if (message instanceof RoleChanged) {
            // this message is sent by RaftActor. Notify registered listeners when this message is received.
            RoleChanged roleChanged = (RoleChanged) message;

            LOG.info("RoleChangeNotifier for {} , received role change from {} to {}", memberId,
                roleChanged.getOldRole(), roleChanged.getNewRole());

            latestRoleChangeNotification =
                new RoleChangeNotification(roleChanged.getMemberId(),
                    roleChanged.getOldRole(), roleChanged.getNewRole());

            for (ActorRef listener: registeredListeners.values()) {
                listener.tell(latestRoleChangeNotification, getSelf());
            }
        }
    }

    @Override
    public void close() throws Exception {
        registeredListeners.clear();
    }
}


@@ -366,33 +366,8 @@
 s;%0A%0A
-import akka.actor.Actor;%0A
 impo
@@ -448,34 +448,8 @@
 ps;%0A
-import akka.japi.Creator;%0A
 impo
@@ -1344,186 +1344,42 @@
 ate(
-new Creator%3CActor%3E() %7B%0A            @Override%0A            public Actor create() throws Exception %7B%0A                return new RoleChangeNotifier(memberId);%0A            %7D%0A        %7D
+RoleChangeNotifier.class, memberId
 );%0A 

Fix Akka exception: cannot use non-static local Creator to create actors