package jforex;

import com.dukascopy.api.*;

import akka.actor.*;
import akka.actor.Actor.*;
import akka.event.*;

import com.typesafe.config.ConfigFactory;

@RequiresFullAccess
@Library("akka-actor-2.0-M2.jar:akka-remote-2.0-M2.jar:netty-3.2.6.Final.jar:protobuf-java-2.4.1.jar:scala-library-2.9.1.jar")

public class JfxPing implements IStrategy {

  public class Client implements Runnable {

    /** Client configuration */
    public static final String akkaConf = "\n"+
      "\n" +
      "ping {\n" +
        "\n" +
        "akka {\n" +
          "\n" +
          "actor {\n" +
            "\n" +
            "provider = \"akka.remote.RemoteActorRefProvider\"\n" +
            "\n" +
          "}\n" +
          "\n" +
          "remote {\n" +
            "\n" +
            "transport = \"akka.remote.netty.NettyRemoteSupport\"\n" +
            "\n" +
            "server {\n" +
              "\n" +
              "hostname = \"127.0.0.1\"\n" +
              "port = 3555\n" +
              "\n" +
            "}\n" +
            "\n" +
          "}\n" +
          "\n" +
          "cluster.nodename = \"pong\"\n" +
          "\n" +
        "}\n" +
      "}\n";

    /** Server configuration */
    public static final String serverPath = "akka://[email protected]:2552/user/pong";

    /** Client playing field */
    public class PingField extends UntypedActor {

      LoggingAdapter log = Logging.getLogger(getContext().system(), this);

      /** Max. game rounds */
      int rounds = -1;

      /** Mailbox */
      @Override
      public void onReceive(Object message) throws Exception {

        // String based messages
        if (message instanceof String) {

          String text = (String) message;

          // Counter
          if(text.equals("PONG")) {

            console.getOut().println(text);

            // Check game over
            if(rounds != -1)
              rounds--;
            if(rounds == 0) {

              // Tell it
              getSender().tell("GAME OVER");
              getSender().tell("SHUTDOWN");

              // Shutdown all
              getContext().stop(getSelf());
              context.stop();

            }

            // A little break
            Thread.sleep(1000);

            // Continue the game
            getSender().tell("PING", getSelf());

          // Unsupported message
          } else
            console.getOut().println("Protocol not support: " + text);

        // Unsupported message
        } else
          console.getOut().println("Protocol not support: " + message);
      }
    }

    public void run() {

      // Read client configuration
      system = ActorSystem.create("Client",
                                  ConfigFactory.parseString(akkaConf).getConfig("ping"));

      // Init client
      ping = system.actorOf(new Props(new UntypedActorFactory() {

        public UntypedActor create() {

          return new PingField();

        }

      } ), "ping");

      // Connect to server
      pong = system.actorFor(serverPath);

      // Send ping
      pong.tell("PING", ping);

    }

  }

  public void onStart(IContext context) throws JFException {

    // Init JForex.
    this.engine = context.getEngine();
    this.console = context.getConsole();
    this.history = context.getHistory();
    this.context = context;
    this.indicators = context.getIndicators();
    this.userInterface = context.getUserInterface();

    Client client= new Client();
    new Thread(client).start();

  }

  public void onAccount(IAccount account) throws JFException {
  }

  public void onMessage(IMessage message) throws JFException {
  }

  public void onStop() throws JFException {
  }

  public void onTick(Instrument instrument, ITick tick) throws JFException {


  }

  public void onBar(Instrument instrument, Period period, IBar askBar, IBar bidBar) throws JFException {
  }


  /** JForex objects. */
  private IEngine engine;
  static private IConsole console;
  private IHistory history;
  private IContext context;
  private IIndicators indicators;
  private IUserInterface userInterface;

  /** Akka objects */

  private ActorSystem system;
  private ActorRef ping;
  private ActorRef pong;

}