/*!
        Functions are specific to the assisi message protocol,
        i.e., each message is a multipart message in the format:
        
        +---------+
        | header1 |
        +---------+
        | header2 |
        +---------+
        | header3 |
        +---------+
        |  data   |
        +---------+

        The function asserts the above message format.
    */
    static int recv_multipart(socket_t& socket,
                       std::string& name,
                       std::string& device,
                       std::string& command,
                       std::string& data,
                       int flags = 0)
    {
        int len = 0;
        message_t msg;
        
        len = socket.recv(&msg, flags);
        if (len > 0)
        {
            name = msg_to_str(msg);
            //assert(!last_part(socket));
            len += socket.recv(&msg);
            device = msg_to_str(msg);
            //std::cout << device << std::endl;

            //assert(!last_part(socket));
            len += socket.recv(&msg);
            command = msg_to_str(msg);
            //std::cout << command << std::endl;
            
            //assert(!last_part(socket));
            len += socket.recv(&msg);
            data = msg_to_str(msg);
            //std::cout << data << std::endl;
        }
        return len;
    }

int SendEnvelopes(socket_t &socket, vector<string> const &identities, vector<Envelope> &envelopes, bool is_protocol, int flags)
{
    int initial_flags = ZMQ_SNDMORE | (ZMQ_DONTWAIT & flags);

    for (vector<string>::size_type i = 0; i < identities.size(); i++) {
        string identity = identities.at(i);
        if (!socket.send(identity.data(), identity.size(), initial_flags)) {
            return 0;
        }
    }

    message_t msg;
    msg.rebuild(0);
    if (!socket.send(msg, initial_flags)) return -1;

    for (vector<Envelope>::size_type i = 0; i < envelopes.size(); i++) {
        if (is_protocol) {
            if (!envelopes.at(i).ToProtocolZmqMessage(msg)) return -1;
        }
        else {
            if (!envelopes.at(i).ToZmqMessage(msg)) return -1;
        }

        if (!socket.send(msg, flags)) return 0;
    }

    return 1;
}

int SendEnvelope(socket_t &socket, vector<string> const &identities, Envelope &e, bool is_protocol, int flags)
{
    message_t msg;
    int initial_flags = ZMQ_SNDMORE | (ZMQ_DONTWAIT & flags);

    message_t e_msg;
    if (is_protocol) {
        if (!e.ToProtocolZmqMessage(e_msg)) return -1;
    }
    else {
        if (!e.ToZmqMessage(e_msg)) return -1;
    }

    for (size_t i = 0; i < identities.size(); i++) {
        string identity = identities.at(i);
        msg.rebuild(identity.size());
        memcpy(msg.data(), identity.data(), msg.size());
        if (!socket.send(msg, initial_flags)) return 0;
    }

    msg.rebuild(0);
    if (!socket.send(msg, initial_flags)) return 0;

    return socket.send(e_msg, flags) ? 1 : 0;
}

 /*! Returns true if there are no further parts.
  */
 inline bool last_part(socket_t& socket)
 {
     int64_t more;
     size_t more_size = sizeof(more);
     socket.getsockopt(ZMQ_RCVMORE, &more, &more_size);
     return  (more == 1) ? false : true;
 }

bool RequestProcessor::SendSubscriptionEnvelopeResponse(socket_t &sock, EnvelopeSubscriptionResponseState &state)
{
    if (!state.finalized && !(state.current_size > state.max_size || state.messages.size() > state.max_envelopes))
        return true;

    if (state.messages.size() == 0)
        return true;

    Envelope response;
    protocol::response::SubscriptionStartV1 start;
    start.set_id(state.id);
    start.add_to_envelope(response);

    if (!zeromq::SendEnvelope(sock, state.identities, response, true, ZMQ_SNDMORE)) {
        return false;
    }

    vector<message_t *>::iterator i = state.messages.begin();
    for (; i != state.messages.end(); i++) {
        sock.send(**i, i != state.messages.end() ? ZMQ_SNDMORE : 0);
        delete *i;
        *i = NULL;
    }

    state.messages.clear();
    state.current_size = 0;

    return true;
}

std::string _get_zmq_last_endpoint(socket_t & socket) {
    char buf[512];
    size_t buf_size = sizeof(buf);
    //int retval =
    socket.getsockopt(ZMQ_LAST_ENDPOINT, (void*)buf, &buf_size);
    // assert(retval == 0);
    // buf_size is modified.
    return std::string(buf, buf_size);
}

void   clear_pending_recv(socket_t sw)
{
    if((sw)->clear_pending_io)
    {
        lnode *tmp;
        while((tmp = llist_pop(&sw->pending_recv))!=NULL)
            sw->clear_pending_io((st_io*)tmp);
    }
}

 /*! Outgoing messages conform to the sam format as incoming mssages,
     i.e., they have 4 header parts and a data part.
 */
 static int send_multipart(socket_t& socket,
                    const std::string& name,
                    const std::string& device,
                    const std::string& desc,
                    const std::string& data,
                    int flags = 0)
 {
     message_t msg;
     str_to_msg(name, msg);
     socket.send(msg, ZMQ_SNDMORE);
     str_to_msg(device, msg);
     socket.send(msg, ZMQ_SNDMORE);
     str_to_msg(desc, msg);
     socket.send(msg, ZMQ_SNDMORE);
     str_to_msg(data, msg);
     socket.send(msg);
            
     return 4;
 }

void process_connect(socket_t s)
{
    int err = 0;
    socklen_t len = sizeof(err);

    if (getsockopt(s->fd, SOL_SOCKET, SO_ERROR, &err, &len) == -1) {
        s->on_connect(INVALID_SOCK,&s->addr_remote,s->ud,err);
        CloseSocket(s->sock);
        return;
    }

    if(err){
        errno = err;
        s->on_connect(INVALID_SOCK,&s->addr_remote,s->ud,errno);
        CloseSocket(s->sock);
        return;
    }
    //connect success
    s->engine->UnRegister(s->engine,s);
    len = sizeof(s->addr_local);
    getsockname(s->fd,(struct sockaddr*)&s->addr_local,&len);
    s->on_connect(s->sock,&s->addr_remote,s->ud,0);
}

static inline void _send(socket_t s)
{
	assert(s);
	st_io* io_req = 0;
	uint32_t err_code = 0;
	if(s->writeable)
	{
        if((io_req = LLIST_POP(st_io*,&s->pending_send))!=NULL)
		{
			int32_t bytes_transfer = raw_send(s,io_req,&err_code);
			if(bytes_transfer == 0) bytes_transfer = -1;
			if(err_code != EAGAIN)  s->io_finish(bytes_transfer,io_req,err_code);
		}
	}
}

static inline void _recv(socket_t s)
{
	assert(s);
	int32_t ret = -1;
	int32_t bytes_transfer = 0;
	st_io* io_req = 0;
	if(s->readable)
	{
		if(io_req = LINK_LIST_POP(st_io*,s->pending_recv))
		{
			ret = raw_recv(s,io_req,&bytes_transfer,&io_req->err_code);
			if(io_req->err_code != EAGAIN)
				s->OnRead(bytes_transfer,io_req);
		}
	}
}

void process_accept(socket_t s)
{
    SOCK client;
    //struct sockaddr_in ClientAddress;
    socklen_t len = sizeof(s->addr_remote);
    while(1)
    {
        client = _accept(s, (struct sockaddr*)&s->addr_remote,&len);
        if (client == INVALID_SOCK)
            break;
        else{
            socket_t _client = (socket_t)client;
            len = sizeof(_client->addr_local);
            getpeername(_client->fd,(struct sockaddr*)&_client->addr_remote,&len);
            getsockname(_client->fd,(struct sockaddr*)&_client->addr_local,&len);
            s->on_accept(client,&_client->addr_remote,s->ud);
        }
    }
}

string processControl (const string Msg,const socket_t &sock) {
	const CStrings data=split(" ",Msg);
	const CStringsIterator name=data.begin();
	string result="";
	makeLog(LOG_DEBUG,"Control: Processing {%s}",Msg.c_str());

	if (name->compare("SEND")==0) {
		uint16_t resendNumber = (data.size()>1)?atoi(data[1].c_str()):0;
		uint16_t cnt = (data.size()>2)?atoi(data[2].c_str()):0;
		if (cnt<1) cnt=1;
		
//		makeLog(LOG_DEBUG,"RESEND request: %d:%d",resendNumber,cnt);

		bool wasMissing=false;
		for (int i=0;i<cnt;i++) {
			int missing = getDatagramLength(resendNumber);
			
			if (missing<=0) {
				wasMissing=true;
				makeLog(LOG_INFO,"Streamer: Datagram lost: %d",resendNumber);
				sendControl(stringprintf("SEND_OOB %d",resendNumber));
			} else {
				ictDatagram_t ictDatagram;
				getDatagram(ictDatagram,resendNumber);
				
				outQueue.push_front	( resendNumber );
				
//				sendto(sender.socket,ictDatagram,ictDatagram.size(),0,(const sockaddr*)&sender.peer,sizeof(sender.peer));
			}
			
			resendNumber++;
		}

		if (wasMissing) {
			makeLog(LOG_WARNING,"Streamer: Packet(s) is out of buffer");
		}
	}
	else if (name->compare("STARTED")==0) {
		makeLog (LOG_INFO,"Control: method STARTED");
	}
	else if (name->compare("KEPT")==0) {
		keptLastTime=getMNow();
	}
	else if (name->compare("SETBUF")==0) {
		//makelog ("Control message: {$msg1}");
		int newBufferTime=(data.size()>1)?atoi(data[1].c_str()):-1;

		if (newBufferTime!=-1 && cleaningArgs.bufferSize!=newBufferTime) {
			cleaningArgs.bufferSize=newBufferTime;
			makeLog (LOG_INFO,"Buffer is set to new value: %d",cleaningArgs.bufferSize);
		}
		return stringprintf("BUFSET %d",cleaningArgs.bufferSize);
	}
	else if (name->compare("RESETSTREAM")==0) {
		FD_CLR(sender.getHandle(),&master);
		sender.reinit();
		FD_SET(sender.getHandle(),&master);
		
		makeLog (LOG_INFO,"Streamer: Restart on receiver requested");
		result+=stringprintf("STREAMRESET %d",0);
	}
	else if (name->compare("SUSPEND")==0) {
		if (!sendingSuspended) {
			sendingSuspended=true;
			makeLog(LOG_INFO,"Streamer: Supended on receiver request");
		}
		
		return stringprintf("SUSPENDED %d",0); //TODO O by malo byt cislo noveho portu na vysielacej strane
	}
	else if (name->compare("RESUME")==0) {
		sendingSuspended=false;
	
		return stringprintf("RESUMED %d",0); //TODO O by malo byt cislo noveho portu na vysielacej strane
	}
	else if (name->compare("CINIT")==0) {
		int newBufferTime=data.size()>1?atoi(data[1].c_str()):-1;
		sendingSuspended=false;
		makeLog(LOG_INFO,"Clinet initialization request from receiver");

		if (newBufferTime!=-1 && cleaningArgs.bufferSize!=newBufferTime) {
			cleaningArgs.bufferSize=newBufferTime;
			makeLog (LOG_INFO,"Buffer is set to new value: %d",cleaningArgs.bufferSize);
			result+=stringprintf("BUFSET %d",cleaningArgs.bufferSize);
		}

		result+=stringprintf("CINITED %d",0);//getSocketLocalPort(sender.socket)); //TODO O by malo byt cislo noveho portu na vysielacej strane
	}

	else if (name->compare("CUNINIT")==0) {
		clearBuffer();
		sendingInitialized=false;
		result+=stringprintf("CUNINITED %d",0); //TODO O by malo byt cislo noveho portu na vysielacej strane
	}
	else if (name->compare("CONNECTCONTROL")==0) {
		return "CONTROLCONNECT\n";
	} else {
		makeLog(LOG_ERR,"Control: unknown command: %s",name->c_str());
	}

	return result;
//.........这里部分代码省略.........

        bool send(const std::string& route,
                  const packed<Domain, Command>& command)
        {
            const bool multipart = boost::tuples::length<
                typename packed<Domain, Command>::tuple_type
            >::value;

            return send(protect(route), ZMQ_SNDMORE) &&
                   send(static_cast<int>(Command), multipart ? ZMQ_SNDMORE : 0) &&
                   send_multipart(command);
        }

void *controlThreadFcn (void *argPtr) {
	makeLog(LOG_DEBUG,"Control: Thread started");
	ControlArgs *args=(ControlArgs *)argPtr;

	int fdmax=-1;
	fd_set read_fds;
	int hSocket;
	timeval timeout={10,0};//sec,usec

	FD_SET(control.getHandle(), &controlSocks);
	control.updatefdmax(&fdmax);
	FD_SET(sender .getHandle(), &controlSocks);
	sender.updatefdmax(&fdmax);

	for (;;) {
		read_fds=controlSocks;
		int selectCnt=select(fdmax+1, &read_fds, NULL, NULL, &timeout);

		if(selectCnt == -1) {
			makeLog(LOG_ERR,"Control: select error: %s",strerror(errno));
			terminate();
		}
		else if(selectCnt > 0) {
			for(hSocket = 0; hSocket <= fdmax; hSocket++) {
				if(FD_ISSET(hSocket, &read_fds))
				{ // we got one...
					makeLog(LOG_DEBUG,"Processing control from thread");
					if(control.handles(hSocket)) {
						char cmsg[1500];
						int recvlen = control.recv((void*)cmsg,sizeof(cmsg),0);

						if (recvlen>0) {
							controlStr.append(cmsg,recvlen);
							processControls(controlStr,control);
						} else if (recvlen==0) {
							FD_CLR( hSocket, &controlSocks );
						}
					}

					if (sender.handles(hSocket)) {
						char cmsg[1500];
						int recvlen = sender.recv(cmsg,sizeof(cmsg),0);

						if (recvlen>0) {
							controlStr.append(cmsg,recvlen);
							processControls(controlStr,sender);
						} else if (recvlen==0) {
							FD_CLR( hSocket, &controlSocks );
						}
					}
				}//if FD_ISSET
			} //for loop through sockets
		} //selectCnt > 0
	}//infinite loop
	
	FD_CLR( control.getHandle(), &controlSocks );
	FD_CLR( sender .getHandle(), &controlSocks );
}

        void drop() {
            zmq::message_t null;

            while(more()) {
                recv(&null);
            }
        }

        bool recv(T& result,
                  int flags = 0)
        {
            zmq::message_t message;
            msgpack::unpacked unpacked;

            if(!recv(&message, flags)) {
                return false;
            }
           
            try { 
                msgpack::unpack(
                    &unpacked,
                    static_cast<const char*>(message.data()),
                    message.size()
                );
                
                unpacked.get().convert(&result);
            } catch(const msgpack::type_error& e) {
                throw std::runtime_error("corrupted object");
            } catch(const std::bad_cast& e) {
                throw std::runtime_error("corrupted object - type mismatch");
            }

            return true;
        }

bool sendControl(const string data) {
	makeLog(LOG_DEBUG,"Control: Sending {%s}",data.c_str());

	if (control.peer.empty()) {
	    makeLog(LOG_ERR,"Control: Peer party is not connected!");
	    return false;
	}

	string buf(data);
	buf.append("\n");
	int sentLength=control.write(buf);

	if (sentLength <= 0) {
		return false;
	} else if (buf.size() != sentLength) {
		makeLog(LOG_ERR,"Control: Unable to send complete data");
	}

	return true;

}

int main(int argc, char *argv[]) {
	//clear the master and temp sets
	FD_ZERO(&master);
	FD_ZERO(&controlSocks);
	
	char c;
	int hSocket;
	uint16_t number=0;
	uint64_t lastAnnounceTime=0;
	Reconnector reconnector;

	set_terminate(terminateFcn);

	void (*prev_fn)(int);
	prev_fn = signal (SIGINT,terminateFcn);
	if (prev_fn==SIG_ERR)
		makeLog(LOG_ERR,"Main: Unable to set termination funtion"); 

	while ((c = getopt (argc, argv, "fvc:")) != -1)
	switch (c) {
	    case 'f':
		fflag = true;
	    break;
	    case 'v':
		vflag = true;
	    break;
	    case 'c':
		configFile = optarg;
	    break;
	    case '?':
		if (optopt == 'c')
		    fprintf (stderr, "Main: Option -%c requires an argument.\n", optopt);
		else if (isprint (optopt))
		    fprintf (stderr, "Main: Unknown option `-%c'.\n", optopt);
		else
		    fprintf (stderr, "Main: Unknown option character `\\x%x'.\n", optopt);
		return 1;
	    default:
		abort ();
	}

	Config config;
	readConfig(config);
	makeLog(LOG_DEBUG,"Read configuration");
	
	if (!pidFile.empty()) {
		makeLog(LOG_DEBUG,"Main: Checking pid file: %s",pidFile.c_str());
		ifstream pidfile(pidFile.c_str());
		if (pidfile.is_open()) {
			char line[20];
			int otherpid;
			pidfile.getline(line,sizeof(line));
			sscanf(line,"%u",&otherpid);
			
			if (kill(otherpid,0)==0) {
				makeLog(LOG_ERR,"Main: Other instance is running: %d",otherpid);
				pidFile="";
				terminate();
			}
			pidfile.close();
		}
	}
	
//	makeLog(LOG_DEBUG,"PID check finished");
	
	if (!fflag) daemon (0,1);
	
	if (!pidFile.empty()) {
		ofstream pidfile;

		pidfile.open(pidFile.c_str());
		pidfile << getpid();
		pidfile << "\n";
		pidfile.close();
	}

	if (listeningArgs.proto.compare("tcp")==0) {
		//get the listener
		if(listener.init(AF_INET, SOCK_STREAM, 0) == -1) {
			makeLog(LOG_ERR,"Listener: Unable to create socket");
			exit(1);
		}
		
		listener.peer.set_in(listeningArgs.host,listeningArgs.port);

		if (listener.connect() < 0) {
			makeLog(LOG_ERR,"Listener: Cannot connect: %s;",strerror(errno));
			reconnector.add(&listener,reconnectListener,NULL);
		} else {
			// add the listener to the master set
			listener.updatefdmax(&fdmax);
			makeLog(LOG_INFO,"Listener: Connected to %s:%d;",listeningArgs.host.c_str(),listeningArgs.port);
		}

		FD_SET(listener.getHandle(), &master);
	} else {
		const bool mcast=((inet_addr(listeningArgs.host.c_str()) & MCAST_MASK) == MCAST_ADDR);

		//get the listener
		if(listener.init(AF_INET, SOCK_DGRAM, 0) == -1) {
//.........这里部分代码省略.........