#include "reqresp/req-dev.h"
#include "utils/type.h"
#include "triple.h"
using namespace std;

//for resources with out-degree 0, we treat it like a value of the edge src

#define NO_RESP_REPORT false

int count_not_matched = 0;//+++++++++++++++

//splitter: \t
//input line format: vid context subject nb1_predicate nb1_obj nb2_predicate nb2_obj ...
//output line format: vid context subject nb1 nb1_predicate nb2 nb2_predicate ... (for a literal field, nb is the value field)

struct S2IValue_req
{
	int vid;
	string context;
	vector<string> nb;
	vector<bool> nb_type;//is_literal = true
	vector<string> elabel;
	vector<VertexID> nb_id;
};

ibinstream & operator<<(ibinstream & m, const S2IValue_req & v){
	m<<v.vid;
	m<<v.context;
	m<<v.nb;
	m<<v.nb_type;
	m<<v.elabel;
	return m;
}

obinstream & operator>>(obinstream & m, S2IValue_req & v){
	m>>v.vid;
	m>>v.context;
	m>>v.nb;
	m>>v.nb_type;
	m>>v.elabel;
	return m;
}

//====================================

class S2IVertex_req: public RVertex<string, S2IValue_req, char, VertexID, StringHash> {
	public:
		virtual int respond()
		{
			return value().vid;
		}

		virtual void compute(MessageContainer & messages) {
			vector<string> & nb=value().nb;
			vector<bool> & nb_type=value().nb_type;
			if (step_num() == 1) {
				for(int i=0; i<nb.size(); i++)
				{
					if(nb_type[i]) continue;
					string tgt = value().context+"\t"+nb[i];
					request(tgt);
				}
			}
			else
			{	vector<VertexID> & nb_id = value().nb_id;
				nb_id.resize(nb.size());//very important !!!
				vector<bool> & nb_type=value().nb_type;
				for(int i=0; i<nb.size(); i++)
				{
					if(nb_type[i]) continue;
					string tgt = value().context+"\t"+nb[i];
					int * resp = get_respond_safe(tgt);
					if(resp == NULL)
					{
						//for resources with out-degree 0, we treat it like a value of the edge src
						nb_type[i] = true;
						count_not_matched++;//+++++++++++++++
					}
					else nb_id[i] = *resp;
				}
				vote_to_halt();
			}
		}
};

class S2IWorker_req: public RWorker<S2IVertex_req> {
	char* buf;

	public:

		S2IWorker_req()
		{
			buf=new char[TRIPLE_STRBUF_SIZE];
		}

		~S2IWorker_req()
		{
			delete buf;
		}

		virtual S2IVertex_req * toVertex(char* line) {
			statement o;
			const char* l=line;
			o.skipWhiteSpace(l);
			S2IVertex_req * v = new S2IVertex_req;
			v->value().vid = o.readNum(l);
			o.skipWhiteSpace(l);
			v->value().context = o.readResource(l);
			o.skipWhiteSpace(l);
			v->id = v->value().context;
			v->id += "\t";
			v->id += o.readResource(l);
			o.skipWhiteSpace(l);
			while(!o.eol(l))
			{
				string predicate = o.readResource(l);
				v->value().elabel.push_back(predicate);
				o.skipWhiteSpace(l);
				string object = o.readNode(l);
				v->value().nb.push_back(object);
				if(literal_read) v->value().nb_type.push_back(true);
				else v->value().nb_type.push_back(false);
				o.skipWhiteSpace(l);
			}
			return v;
		}

		virtual void toline(S2IVertex_req * v, BufferedWriter & writer) {
			sprintf(buf, "%d\t%s", v->value().vid, v->id.c_str()); //v->id = context \t subject
			writer.write(buf);
			vector<string> & nb = v->value().nb;
			vector<VertexID> & nb_id = v->value().nb_id;
			vector<string> & elabel = v->value().elabel;
			vector<bool> & nb_type = v->value().nb_type;
			for(int i=0; i<elabel.size(); i++)
			{
				if(nb_type[i]) sprintf(buf, "\t%s\t%s", nb[i].c_str(), elabel[i].c_str());
				else sprintf(buf, "\t%d\t%s", nb_id[i], elabel[i].c_str());
				writer.write(buf);
			}
			writer.write("\n");
		}
};

void req_str2int(string in_path, string out_path) {
	WorkerParams param;
	param.input_path = in_path;
	param.output_path = out_path;
	param.force_write = true;
	param.native_dispatcher = false;
	S2IWorker_req worker;
	worker.run(param);
}

int main(int argc, char* argv[]){
	init_workers();
	req_str2int("/dbpedia_out", "/dbpedia_final");
	int num = master_sum_LL(count_not_matched);//+++++++++++++++
	if(_my_rank == MASTER_RANK) cout<<"Total # of edges with unmatched dst-resource: "<<num<<endl;//+++++++++++++++
	worker_finalize();
	return 0;
}