/*
 * Copyright (c) 2015 NetIDE Consortium 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 net.floodlightcontroller.interceptor;

import eu.netide.lib.netip.NetIPConverter;
import eu.netide.lib.netip.OpenFlowMessage;
import io.netty.buffer.Unpooled;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;
import org.zeromq.ZMsg;

/**
 * @author giuseppex.petralia@intel.com
 *
 */
public class ZeroMQBaseConnector implements Runnable{
    private static final String STOP_COMMAND = "Control.STOP";
    private static final String CONTROL_ADDRESS = "inproc://ShimControllerQueue";

    private static final Logger LOG = LoggerFactory.getLogger(ZeroMQBaseConnector.class);
    private String address;
    private int port;
    private ZMQ.Context context;
    private Thread thread;

    private CoreListener coreListener;

    public ZeroMQBaseConnector() {

    }

    public void Start() {
        context = ZMQ.context(1);
        thread = new Thread(this);
        thread.setName("ZeroMQBasedConnector Receive Loop");
        thread.start();
    }

    public void Stop() {
        if (thread != null) {
            ZMQ.Socket stopSocket = context.socket(ZMQ.PUSH);
            stopSocket.connect(CONTROL_ADDRESS);
            stopSocket.send(STOP_COMMAND);
            stopSocket.close();
            try {
                thread.join();
                context.term();
            } catch (InterruptedException e) {
                LOG.error("", e);
            }
        }
    }

    public void RegisterCoreListener(ICoreListener listener) {
        this.coreListener = listener;
    }

    public boolean SendData(byte[] data) {
        ZMsg msg = new ZMsg();
        msg.add(data);
        ZMQ.Socket sendSocket = context.socket(ZMQ.PUSH);
        sendSocket.setIdentity("shim".getBytes());
        sendSocket.connect(CONTROL_ADDRESS);
        msg.send(sendSocket);
        sendSocket.close();
        return true;
    }

    @Override
    public void run() {
        ZMQ.Socket socket = context.socket(ZMQ.DEALER);
        socket.setIdentity("shim".getBytes());
        socket.connect("tcp://" + getAddress() + ":" + getPort());
        LOG.info("Trying to connect to core on address tcp://" + getAddress() + ":" + getPort());

        ZMQ.Socket controlSocket = context.socket(ZMQ.PULL);
        controlSocket.bind(CONTROL_ADDRESS);

        ZMQ.Poller poller = new ZMQ.Poller(2);
        poller.register(socket, ZMQ.Poller.POLLIN);
        poller.register(controlSocket, ZMQ.Poller.POLLIN);

        while (!Thread.currentThread().isInterrupted()) {
            poller.poll(10);
            if (poller.pollin(0)) {
                ZMsg message = ZMsg.recvMsg(socket);
                byte[] data = message.getLast().getData();
                if (coreListener != null) {
                    Message msg = NetIPConverter.parseConcreteMessage(data);
                    if (msg instanceof HelloMessage) {

                        coreListener.onHelloCoreMessage(((HelloMessage) msg).getSupportedProtocols(),
                                ((HelloMessage) msg).getHeader().getModuleId());
                    } else if (msg instanceof OpenFlowMessage) {

                        byte[] payload = msg.getPayload();
                        coreListener.onOpenFlowCoreMessage(msg.getHeader().getDatapathId(),
                                Unpooled.wrappedBuffer(payload), msg.getHeader().getModuleId());
                    } else {
                        
                    }
                }
            }c
            if (poller.pollin(1)) {
                ZMsg message = ZMsg.recvMsg(controlSocket);
                if (message.getFirst().toString().equals(STOP_COMMAND)) {
                    break;
                } else {
                    message.send(socket);
                }
            }
        }
        socket.close();
        controlSocket.close();
    }

    public void setPort(int port) {
        this.port = port;
    }

    public int getPort() {
        return port;
    }

    public void setAddress(String address) {
        this.address = address;
    }

    public String getAddress() {
        return address;
    }

}

@@ -346,16 +346,93 @@
 eptor;%0A%0A
+import eu.netide.lib.netip.HelloMessage;%0Aimport eu.netide.lib.netip.Message;%0A
 import e
@@ -751,16 +751,17 @@
 Runnable
+ 
 %7B%0A    pr
@@ -1842,17 +1842,16 @@
 istener(
-I
 CoreList
@@ -3764,32 +3764,8 @@
 e %7B%0A
-                        
 %0A   
@@ -3814,17 +3814,16 @@
        %7D
-c
 %0A       

Fix Core Connector imports