// Path: src/main/java/com/sulaco/fuse/akka/actor/FuseEndpointActor.java
// public abstract class FuseEndpointActor extends FuseBaseActor {
//     
//     @Autowired protected MetricsRegistry metrics;
//     
//     public FuseEndpointActor() {
//         super();
//     }
// 
//     // TODO: refactor metrics
//     //            meter = metrics.getRegistry().timer(getClass().getName());
// 
//     @Override
//     public void onReceive(Object message) throws Exception {
//         try {
//             if (message instanceof FuseRequestMessage) {
//                 onRequest((FuseRequestMessage) message);
//                 //proto.error((FuseRequestMessage) message);
//             }
//             else {
//                 super.onReceive(message);
//             }
//         }
//         catch (Exception ex) {
//             log.error("Error handling request !", ex);
//             unhandled(message);
//         }
//         finally {
//             // TODO: context flush
//         }
//     }
//     
//     protected void onRequest(final FuseRequestMessage request) {
//         
//         RouteHandler rhandler = request.getHandler();
//         
//         Optional<String> method = rhandler.getMethodName();
//         
//         if (method.isPresent()) {
//             // Invoke configured method instead of default 'onReceive'. Method needs to have correct
//             // signature, otherwise, fallback to 'onReceive' will occur.
//             //
//             Optional<Method> target = lookupMethod(this, method.get());
//             if (target.isPresent()) {
//                 try {
//                     target.get()
//                           .invoke(this, request);
//                 }
//                 catch (Exception ex) {
//                     log.warn("[fuse] Invocation failure. x_x", ex);
//                 }
//             }       
//         }
//         else {
//             log.warn("[fuse] No handling method specified. Override onReceive. x_x");
//         }
//     }
// 
// 
// 
//     @Override
//     public void unhandled(Object message) {
//         super.unhandled(message);
//         if (message instanceof FuseRequestMessage) {
//             proto.error((FuseRequestMessage) message);
//         }
//     }
// 
//     public void setProto(WireProtocol proto) {
//         this.proto = proto;
//     }
// 
//     protected static final Logger log = LoggerFactory.getLogger(FuseEndpointActor.class);
// }
// 
// Path: src/main/java/com/sulaco/fuse/akka/message/FuseInternalMessage.java
// public interface FuseInternalMessage extends FuseOriginChain {
//     
//     FuseMessageContext getContext();
// 
//     long getRequestId();
// 
//     long getTimestamp();
// 
//     FuseInternalMessage timestamp();
// }
// 
// Path: src/main/java/com/sulaco/fuse/akka/message/FuseRequestMessage.java
// public interface FuseRequestMessage extends Route {
// 
//     long getId();
// 
//     HttpRequest getRequest();
// 
//     String getRequestBody();
// 
//     ChannelHandlerContext getChannelContext();
// 
//     void flush();
// 
//     boolean flushed();
// 
//     public <T> Optional<T> getParam(String name, Class<T> clazz);
// 
// }
// 
// Path: examples/async/src/main/java/com/sulaco/fuse/example/async/domain/dao/CassandraDao.java
// public interface CassandraDao {
// 
//     public void getSongById(String id, Function<Optional<?>, ?> fuseConsumer);
// 
//     public void getPlaylistById(String id, Function<Optional<?>, ?> fuseConsumer);
// }
// 
// Path: examples/async/src/main/java/com/sulaco/fuse/example/async/domain/entity/Playlist.java
// public class Playlist {
// 
//     UUID id;
//     Set<Song> songs;
// 
//     public Playlist() {
//         this.songs = new HashSet<>();
//     }
// 
//     public Playlist(UUID id) {
//         this();
//         this.id = id;
//     }
// 
//     public UUID getId() {
//         return id;
//     }
// 
//     public Set<Song> getSongs() {
//         return songs;
//     }
// 
//     public void addSong(Song song) {
//         this.songs.add(song);
//     }
// 
//     @Override
//     public boolean equals(Object o) {
//         if (this == o) return true;
//         if (!(o instanceof Playlist)) return false;
// 
//         Playlist playlist = (Playlist) o;
// 
//         if (!id.equals(playlist.id)) return false;
// 
//         return true;
//     }
// 
//     @Override
//     public int hashCode() {
//         return id != null ? id.hashCode() : 0;
//     }
// 
//     public static final Playlist EMPTY = new Playlist();
// }


import com.sulaco.fuse.akka.actor.FuseEndpointActor;
import com.sulaco.fuse.akka.message.FuseInternalMessage;
import com.sulaco.fuse.akka.message.FuseRequestMessage;
import com.sulaco.fuse.example.async.domain.dao.CassandraDao;
import com.sulaco.fuse.example.async.domain.entity.Playlist;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import java.util.Optional;
package com.sulaco.fuse.example.async.endpoint;



public class PlaylistReadActor extends FuseEndpointActor {

    @Autowired CassandraDao dao;

    @Override
    protected void onRequest(final FuseRequestMessage request) {

        Optional<String> id = request.getParam("playlistId");

        if (id.isPresent()) {

            // Once suspended request is revived, it will be passed towards an actor
            // represented by custom path. If suspended without a path, this actor is treated
            // as origin, and revival message will be sent back to it. Take a look at FuseBaseActor.onMessage
            //
            suspend(request);

            // we will cross thread boundaries inside the dao, take a look
            dao.getPlaylistById(
                id.get(),
                result -> revive(request, result)
            );
        }
        else {
// Path: src/main/java/com/sulaco/fuse/akka/actor/FuseEndpointActor.java
// public abstract class FuseEndpointActor extends FuseBaseActor {
//     
//     @Autowired protected MetricsRegistry metrics;
//     
//     public FuseEndpointActor() {
//         super();
//     }
// 
//     // TODO: refactor metrics
//     //            meter = metrics.getRegistry().timer(getClass().getName());
// 
//     @Override
//     public void onReceive(Object message) throws Exception {
//         try {
//             if (message instanceof FuseRequestMessage) {
//                 onRequest((FuseRequestMessage) message);
//                 //proto.error((FuseRequestMessage) message);
//             }
//             else {
//                 super.onReceive(message);
//             }
//         }
//         catch (Exception ex) {
//             log.error("Error handling request !", ex);
//             unhandled(message);
//         }
//         finally {
//             // TODO: context flush
//         }
//     }
//     
//     protected void onRequest(final FuseRequestMessage request) {
//         
//         RouteHandler rhandler = request.getHandler();
//         
//         Optional<String> method = rhandler.getMethodName();
//         
//         if (method.isPresent()) {
//             // Invoke configured method instead of default 'onReceive'. Method needs to have correct
//             // signature, otherwise, fallback to 'onReceive' will occur.
//             //
//             Optional<Method> target = lookupMethod(this, method.get());
//             if (target.isPresent()) {
//                 try {
//                     target.get()
//                           .invoke(this, request);
//                 }
//                 catch (Exception ex) {
//                     log.warn("[fuse] Invocation failure. x_x", ex);
//                 }
//             }       
//         }
//         else {
//             log.warn("[fuse] No handling method specified. Override onReceive. x_x");
//         }
//     }
// 
// 
// 
//     @Override
//     public void unhandled(Object message) {
//         super.unhandled(message);
//         if (message instanceof FuseRequestMessage) {
//             proto.error((FuseRequestMessage) message);
//         }
//     }
// 
//     public void setProto(WireProtocol proto) {
//         this.proto = proto;
//     }
// 
//     protected static final Logger log = LoggerFactory.getLogger(FuseEndpointActor.class);
// }
// 
// Path: src/main/java/com/sulaco/fuse/akka/message/FuseInternalMessage.java
// public interface FuseInternalMessage extends FuseOriginChain {
//     
//     FuseMessageContext getContext();
// 
//     long getRequestId();
// 
//     long getTimestamp();
// 
//     FuseInternalMessage timestamp();
// }
// 
// Path: src/main/java/com/sulaco/fuse/akka/message/FuseRequestMessage.java
// public interface FuseRequestMessage extends Route {
// 
//     long getId();
// 
//     HttpRequest getRequest();
// 
//     String getRequestBody();
// 
//     ChannelHandlerContext getChannelContext();
// 
//     void flush();
// 
//     boolean flushed();
// 
//     public <T> Optional<T> getParam(String name, Class<T> clazz);
// 
// }
// 
// Path: examples/async/src/main/java/com/sulaco/fuse/example/async/domain/dao/CassandraDao.java
// public interface CassandraDao {
// 
//     public void getSongById(String id, Function<Optional<?>, ?> fuseConsumer);
// 
//     public void getPlaylistById(String id, Function<Optional<?>, ?> fuseConsumer);
// }
// 
// Path: examples/async/src/main/java/com/sulaco/fuse/example/async/domain/entity/Playlist.java
// public class Playlist {
// 
//     UUID id;
//     Set<Song> songs;
// 
//     public Playlist() {
//         this.songs = new HashSet<>();
//     }
// 
//     public Playlist(UUID id) {
//         this();
//         this.id = id;
//     }
// 
//     public UUID getId() {
//         return id;
//     }
// 
//     public Set<Song> getSongs() {
//         return songs;
//     }
// 
//     public void addSong(Song song) {
//         this.songs.add(song);
//     }
// 
//     @Override
//     public boolean equals(Object o) {
//         if (this == o) return true;
//         if (!(o instanceof Playlist)) return false;
// 
//         Playlist playlist = (Playlist) o;
// 
//         if (!id.equals(playlist.id)) return false;
// 
//         return true;
//     }
// 
//     @Override
//     public int hashCode() {
//         return id != null ? id.hashCode() : 0;
//     }
// 
//     public static final Playlist EMPTY = new Playlist();
// }


// Path: examples/async/src/main/java/com/sulaco/fuse/example/async/endpoint/PlaylistReadActor.java
import com.sulaco.fuse.akka.actor.FuseEndpointActor;
import com.sulaco.fuse.akka.message.FuseInternalMessage;
import com.sulaco.fuse.akka.message.FuseRequestMessage;
import com.sulaco.fuse.example.async.domain.dao.CassandraDao;
import com.sulaco.fuse.example.async.domain.entity.Playlist;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import java.util.Optional;

package com.sulaco.fuse.example.async.endpoint;



public class PlaylistReadActor extends FuseEndpointActor {

    @Autowired CassandraDao dao;

    @Override
    protected void onRequest(final FuseRequestMessage request) {

        Optional<String> id = request.getParam("playlistId");

        if (id.isPresent()) {

            // Once suspended request is revived, it will be passed towards an actor
            // represented by custom path. If suspended without a path, this actor is treated
            // as origin, and revival message will be sent back to it. Take a look at FuseBaseActor.onMessage
            //
            suspend(request);

            // we will cross thread boundaries inside the dao, take a look
            dao.getPlaylistById(
                id.get(),
                result -> revive(request, result)
            );
        }
        else {
            proto.respond(request, Playlist.EMPTY);