returnedrows.cpp

/* Copyright (C) 2013 Calpont Corp.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License along
   with this program; if not, write to the Free Software Foundation, Inc.,
   51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */

#include <unistd.h>
#include <iostream>
#include <string>
#include <cerrno>
using namespace std;

#include <boost/scoped_ptr.hpp>
#include <boost/scoped_array.hpp>
using namespace boost;

#ifndef _MSC_VER
#include "config.h"
#endif

#include "socktype.h"

#include "exceptclasses.h"

#include "socketio.h"

#include "bytestream.h"
#include "messagequeue.h"
using namespace messageqcpp;

#include "rowgroup.h"
using namespace rowgroup;

namespace qfe
{

void processReturnedRows(MessageQueueClient* mqc, SockType fd)
{
	scoped_ptr<MessageQueueClient> cleaner(mqc);
	SBS sbs;
	sbs = mqc->read();
	//cerr << "got a bs of " << sbs->length() << " bytes" << endl;

	RowGroup rg;
	rg.deserialize(*sbs);

	//cerr << "got a base rowgroup with rows of " << rg.getRowSize() << " bytes" << endl;
	//cerr << rg.toString() << endl;

	ByteStream bs;
	ByteStream::quadbyte tableOID=100;
	bs.reset();
	bs << tableOID;
	mqc->write(bs);

	sbs = mqc->read();
	//cerr << "got a bs of " << sbs->length() << " bytes" << endl;
	RGData rgd;
	rgd.deserialize(*sbs, true);
	rg.setData(&rgd);
	//cerr << "got a rowgroup with: " << rg.getRowCount() << " rows" << endl;

	socketio::writeString(fd, "OK");
	Row r;
	while (rg.getRowCount() > 0)
	{
		rg.initRow(&r);
		rg.getRow(0, &r);
		string csv;
		bs.reset();
		for (unsigned i = 0; i < rg.getRowCount(); i++)
		{
			csv = r.toCSV();
			bs << csv;
			r.nextRow();
		}
		//cerr << "writing " << bs.length() << " bytes back to client" << endl;
		SockWriteFcn(fd, bs.buf(), bs.length());

		bs.reset();
		bs << tableOID;
		mqc->write(bs);

		sbs = mqc->read();
		//cerr << "got a bs of " << sbs->length() << " bytes" << endl;
		rgd.deserialize(*sbs, true);
		rg.setData(&rgd);
		//cerr << "got a rowgroup with: " << rg.getRowCount() << " rows" << endl;
	}

	tableOID=0;
	bs.reset();
	bs << tableOID;
	mqc->write(bs);

	//sync with the client on end-of-results
	SockWriteFcn(fd, &tableOID, 4);
	SockReadFcn(fd, &tableOID, 4);

}

}