void reception_thread::read_packet(boost::shared_array<unsigned char> buffer, unsigned int size)
{
  std::cout << "read_packet " << size << std::endl;
  assert(!!boost::get<frequency_lock_state>(&state)
         || !!boost::get<ready_state>(&state));
  if(boost::get<frequency_lock_state>(&state))
  {
  frequency_lock_state& s = boost::get<frequency_lock_state>(state);

  if(size >= 4)
  {
    typedef gts::sections::extended_section<unsigned char*> section_type;
          
    section_type section(buffer.get (), buffer.get () + size);

    typedef section_type::table_id_iterator table_id_iterator;
    table_id_iterator table_id = section.begin ();
    std::cout << "table_id: 0x" << std::hex << *table_id << std::dec << std::endl;
    if(*table_id == 0x42) // SDT
    {
      std::cout << "SDT" << std::endl;
      if(parse_sdt(buffer.get(), size))
      {
        Q_EMIT lock_end_signal ();
      }
      else
        Q_EMIT lock_error_signal ();
    }
  }    
  }
  else
  {
    std::cout << "Already in ready state" << std::endl;
  }
}

inline bool GetMsgParam(const std::string& netmsgbus_msgcontent, boost::shared_array<char>& msgparam, uint32_t& param_len)
{
    std::string msgparamstr;
    if (netmsgbus_msgcontent.length() < SENDER_LEN_BYTES)
      return false;
    uint8_t msgsender_len = (uint8_t)netmsgbus_msgcontent[0];
    if (netmsgbus_msgcontent.length() < SENDER_LEN_BYTES + (std::size_t)msgsender_len + MSGKEY_LEN_BYTES)
      return false;
    uint8_t msgkey_len = (uint8_t)(netmsgbus_msgcontent[SENDER_LEN_BYTES + msgsender_len]);
    uint32_t msgparam_offset = SENDER_LEN_BYTES + msgsender_len + MSGKEY_LEN_BYTES + msgkey_len;
    if (netmsgbus_msgcontent.length() <  msgparam_offset + MSGVALUE_LEN_BYTES)
      return false;

    uint32_t netparam_len = *((uint32_t *)&netmsgbus_msgcontent[msgparam_offset]);
    param_len = ntohl(netparam_len);
    if (netmsgbus_msgcontent.length() < msgparam_offset + MSGVALUE_LEN_BYTES + param_len)
      return false;

    msgparam.reset(new char[param_len]);
    memcpy(msgparam.get(), &netmsgbus_msgcontent[msgparam_offset + MSGVALUE_LEN_BYTES], param_len);
    return true;

    //if(GetMsgKey(netmsgbus_msgcontent, msgparam_str, msgparamstr))
    //{
    //    assert(msgparamstr.size());
    //    msgparam.reset(new char[msgparamstr.size()]);
    //    memcpy(msgparam.get(), msgparamstr.data(), msgparamstr.size());
    //    param_len = msgparamstr.size();
    //    return true;
    //}
    //printf("netmsgbus_msgcontent error, no msgparam found.\n");
    //return false;
}

void test_write_then_read(const boost::shared_array<uint8_t> buf, uint32_t buf_len) {
  boost::shared_ptr<TMemoryBuffer> membuf(new TMemoryBuffer());
  boost::shared_ptr<TZlibTransport> zlib_trans(new TZlibTransport(membuf));
  zlib_trans->write(buf.get(), buf_len);
  zlib_trans->finish();

  boost::shared_array<uint8_t> mirror(new uint8_t[buf_len]);
  uint32_t got = zlib_trans->readAll(mirror.get(), buf_len);
  BOOST_REQUIRE_EQUAL(got, buf_len);
  BOOST_CHECK_EQUAL(memcmp(mirror.get(), buf.get(), buf_len), 0);
  zlib_trans->verifyChecksum();
}

/**
 * Process an IPFIX Packet
 * @return 0 on success
 */
int IpfixParser::processIpfixPacket(boost::shared_array<uint8_t> message, uint16_t length, boost::shared_ptr<IpfixRecord::SourceID> sourceId) {
	IpfixHeader* header = (IpfixHeader*)message.get();
    sourceId->observationDomainId = ntohl(header->observationDomainId);

	if (ntohs(header->length) != length) {
		msg(MSG_ERROR, "IpfixParser: Bad message length - packet length is  %#06x, header length field is %#06x\n", length, ntohs(header->length));
		return -1;
	}

	/* pointer to first set */
	IpfixSetHeader* set = (IpfixSetHeader*)&header->data;

	/* pointer beyond message */
	uint8_t* endOfMessage = (uint8_t*)((uint8_t*)message.get() + length); 

	uint16_t tmpid;
	uint32_t numberOfDataRecords;
	/* while there is space for a set header... */
	while((uint8_t*)(set) + 4 <= endOfMessage) {
		/* check set length */
		if (ntohs(set->length) < 3) {
			msg(MSG_ERROR, "IpfixParser: Invalid set length %u, must be >= 4", ntohs(set->length));
			return -1;
		}

		tmpid=ntohs(set->id);

		switch(tmpid) {
		case IPFIX_SetId_DataTemplate:
			processDataTemplateSet(sourceId, message, set, endOfMessage);
			break;
		case IPFIX_SetId_Template:
			processTemplateSet(sourceId, message, set, endOfMessage);
			break;
		case IPFIX_SetId_OptionsTemplate:
			processOptionsTemplateSet(sourceId, message, set, endOfMessage);
			break;
		default:
			if(tmpid >= IPFIX_SetId_Data_Start) {
				statTotalDRPackets++;
				numberOfDataRecords += processDataSet(sourceId, message, set, endOfMessage);
			} else {
				msg(MSG_ERROR, "processIpfixPacket: Unsupported Set ID - expected 2/3/4/256+, got %d", tmpid);
			}
		}
		set = (IpfixSetHeader*)((uint8_t*)set + ntohs(set->length));
	}
	
	//FIXME: check for out-of-order messages and lost records
	msg(MSG_VDEBUG, "Message contained %u bytes, sequence number was %u", numberOfDataRecords, ntohl(header->sequenceNo));

	return 0;
}

    bool is_robust_coloring(const Graph & g, const int & k,
                            const boost::shared_array<int>& col)
    {
      bool ok = true;
      boost::shared_array<int> tcol(new int[order(g)]);

      /* The coloring is robust if each vertex is locally recolorable. */
      for (int v = 0; v < order(g) && ok; v++) {
        std::copy(col.get(), col.get() + order(g), tcol.get());
        if (!is_locally_recolorable(g, k, tcol, v))
          ok = false;
      }

      return ok;
    }

    // do all the work on 'np' partitions, 'nx' data points each, for 'nt'
    // time steps
    hpx::future<space> do_work(std::size_t np, std::size_t nx, std::size_t nt,
        boost::shared_array<double> data)
    {
        using hpx::dataflow;
        using hpx::util::unwrapped;

        // U[t][i] is the state of position i at time t.
        std::vector<space> U(2);
        for (space& s: U)
            s.resize(np);

        if (!data) {
            // Initial conditions: f(0, i) = i
            std::size_t b = 0;
            auto range = boost::irange(b, np);
            using hpx::parallel::execution::par;
            hpx::parallel::for_each(
                par, boost::begin(range), boost::end(range),
                [&U, nx](std::size_t i)
                {
                    U[0][i] = hpx::make_ready_future(
                        partition_data(nx, double(i)));
                }
            );
        }
        else {
            // Initialize from existing data
            std::size_t b = 0;
            auto range = boost::irange(b, np);
            using hpx::parallel::execution::par;
            hpx::parallel::for_each(
                par, boost::begin(range), boost::end(range),
                [&U, nx, data](std::size_t i)
                {
                    U[0][i] = hpx::make_ready_future(
                        partition_data(nx, data.get()+(i*nx)));
                }
            );
        }

        auto Op = unwrapped(&stepper::heat_part);

        // Actual time step loop
        for (std::size_t t = 0; t != nt; ++t)
        {
            space const& current = U[t % 2];
            space& next = U[(t + 1) % 2];

            for (std::size_t i = 0; i != np; ++i)
            {
                next[i] = dataflow(
                        hpx::launch::async, Op,
                        current[idx(i, -1, np)], current[i], current[idx(i, +1, np)]
                    );
            }
        }

        // Return the solution at time-step 'nt'.
        return hpx::when_all(U[nt % 2]);
    }

void btlauncherAPI::gotDownloadProgram(const FB::JSObjectPtr& callback, 
									   std::wstring& program,
									   std::string& version,
									   bool success,
									   const FB::HeaderMap& headers,
									   const boost::shared_array<uint8_t>& data,
									   const size_t size) {
	TCHAR temppath[500];
	DWORD gettempresult = GetTempPath(500, temppath);
	if (! gettempresult) {
		callback->InvokeAsync("", FB::variant_list_of(false)("GetTempPath")(GetLastError()));
		return;
	}
	std::wstring syspath(temppath);
	syspath.append( program.c_str() );
	boost::uuids::random_generator gen;
	boost::uuids::uuid u = gen();
	syspath.append( _T("_") );
	std::wstring wversion;
	wversion.assign( version.begin(), version.end() );
	syspath.append( wversion );
	syspath.append( _T("_") );
	syspath.append( boost::uuids::to_wstring(u) );
	syspath.append(_T(".exe"));

	HANDLE hFile = CreateFile( syspath.c_str(), GENERIC_WRITE | GENERIC_EXECUTE, FILE_SHARE_READ, NULL, CREATE_ALWAYS, NULL, NULL );
	if (hFile == INVALID_HANDLE_VALUE) {
		
		callback->InvokeAsync("", FB::variant_list_of(false)(GetLastError()));
		return;
	}
	PVOID ptr = (VOID*) data.get();
	DWORD written = 0;
	BOOL RESULT = WriteFile( hFile, ptr, size, &written, NULL);
	CloseHandle(hFile);
	
	if (! RESULT) {
		callback->InvokeAsync("", FB::variant_list_of("FILE")(false)(GetLastError()));
		return;
	}
	std::wstring installcommand = std::wstring(syspath);
	installcommand.append(_T(" /NOINSTALL /MINIMIZED /HIDE"));
	STARTUPINFO info;
	PROCESS_INFORMATION procinfo;
	memset(&info,0,sizeof(info));
	info.cb = sizeof(STARTUPINFO);
	 
	/* CreateProcessW can modify installcommand thus we allocate needed memory */ 
	wchar_t * pwszParam = new wchar_t[installcommand.size() + 1]; 
	const wchar_t* pchrTemp = installcommand.c_str(); 
    wcscpy_s(pwszParam, installcommand.size() + 1, pchrTemp); 

	BOOL bProc = CreateProcess(NULL, pwszParam, NULL, NULL, FALSE, 0, NULL, NULL, &info, &procinfo);
	if(bProc) {
		callback->InvokeAsync("", FB::variant_list_of("PROCESS")(true)(installcommand.c_str())(GetLastError()));
	} else {
		callback->InvokeAsync("", FB::variant_list_of("PROCESS")(false)(installcommand.c_str())(GetLastError()));
	}

}

void TransportPublisherLink::onMessageLength(const ConnectionPtr& conn, const boost::shared_array<uint8_t>& buffer, uint32_t size, bool success)
{
  if (retry_timer_handle_ != -1)
  {
    getInternalTimerManager()->remove(retry_timer_handle_);
    retry_timer_handle_ = -1;
  }

  if (!success)
  {
    if (connection_)
      connection_->read(4, boost::bind(&TransportPublisherLink::onMessageLength, this, _1, _2, _3, _4));
    return;
  }

  ROS_ASSERT(conn == connection_);
  ROS_ASSERT(size == 4);

  uint32_t len = *((uint32_t*)buffer.get());

  if (len > 1000000000)
  {
    ROS_ERROR("a message of over a gigabyte was " \
                "predicted in tcpros. that seems highly " \
                "unlikely, so I'll assume protocol " \
                "synchronization is lost.");
    drop();
    return;
  }

  connection_->read(len, boost::bind(&TransportPublisherLink::onMessage, this, _1, _2, _3, _4));
}

void test_write_after_flush(const boost::shared_array<uint8_t> buf, uint32_t buf_len) {
  // write some data
  boost::shared_ptr<TMemoryBuffer> membuf(new TMemoryBuffer());
  boost::shared_ptr<TZlibTransport> zlib_trans(new TZlibTransport(membuf));
  zlib_trans->write(buf.get(), buf_len);

  // call finish()
  zlib_trans->finish();

  // make sure write() throws an error
  try {
    uint8_t write_buf[] = "a";
    zlib_trans->write(write_buf, 1);
    BOOST_ERROR("write() after finish() did not raise an exception");
  } catch (TTransportException& ex) {
    BOOST_CHECK_EQUAL(ex.getType(), TTransportException::BAD_ARGS);
  }

  // make sure flush() throws an error
  try {
    zlib_trans->flush();
    BOOST_ERROR("flush() after finish() did not raise an exception");
  } catch (TTransportException& ex) {
    BOOST_CHECK_EQUAL(ex.getType(), TTransportException::BAD_ARGS);
  }

  // make sure finish() throws an error
  try {
    zlib_trans->finish();
    BOOST_ERROR("finish() after finish() did not raise an exception");
  } catch (TTransportException& ex) {
    BOOST_CHECK_EQUAL(ex.getType(), TTransportException::BAD_ARGS);
  }
}

  /*
   * return number of bytes received, negative if error, 0 for received nothing,
   *  which should be treated as error
   */
  int32_t RecvMsg(zmq::socket_t &sock, boost::shared_array<uint8_t> &data,
      bool *_more){
    zmq::message_t msgt;
    int nbytes;
    try{
      nbytes = sock.recv(&msgt);
    }catch(zmq::error_t &e){
      LOG(ERROR) << "RecvMsg error = " << e.what();
      return -1;
    }

    if(nbytes == 0) return 0;

    size_t len = msgt.size();    
    uint8_t *dataptr;
    try{
      dataptr = new uint8_t[len];
    }catch(std::bad_alloc &e){
      return -1;
    }
    
    memcpy(dataptr, msgt.data(), len);
    data.reset(dataptr);

    if(_more != NULL){
      int more_recv;
      size_t s = sizeof(int);
      sock.getsockopt(ZMQ_RCVMORE, &more_recv, &s);
      *_more = (more_recv == 1) ? true : false;
    }
    return len;
  }

/*
 * return number of bytes received, negative if error, 0 for received nothing, which should be treated as error
 */
int recv_msg(zmq::socket_t &sock, boost::shared_array<uint8_t> &data)
{
    zmq::message_t msgt;
    int nbytes;
    try{
      nbytes = sock.recv(&msgt);
    }catch(zmq::error_t e){
      //LOG(NOR, stderr, "recv failed: %s\n", e.what());
      return -1;
    }

    if(nbytes == 0) return 0;

    size_t len = msgt.size();    
    uint8_t *dataptr;
    try{
      dataptr = new uint8_t[len];
    }catch(std::bad_alloc e){
      //LOG(DBG, stderr, "can not allocate memory!\n");
      return -1;
    }
    
    memcpy(dataptr, msgt.data(), len);
    data.reset(dataptr);
    return len;
}

void FBTestPluginAPI::getURLCallback(const FB::JSObjectPtr& callback, bool success,
    const FB::HeaderMap& headers, const boost::shared_array<uint8_t>& data, const size_t size) {
    FB::VariantMap outHeaders;
    for (FB::HeaderMap::const_iterator it = headers.begin(); it != headers.end(); ++it) {
        if (headers.count(it->first) > 1) {
            if (outHeaders.find(it->first) != outHeaders.end()) {
                outHeaders[it->first].cast<FB::VariantList>().push_back(it->second);
            } else {
                outHeaders[it->first] = FB::VariantList(FB::variant_list_of(it->second));
            }
        } else {
            outHeaders[it->first] = it->second;
        }
    }
    if (success) {
        std::string dstr(reinterpret_cast<const char*>(data.get()), size);
        callback->InvokeAsync("", FB::variant_list_of
            (true)
            (outHeaders)
            (dstr)
            );
    } else {
        callback->InvokeAsync("", FB::variant_list_of(false));
    }
}

// 0 for received nothing
int recv_msg_async(zmq::socket_t &sock, boost::shared_array<uint8_t> &data)
{
    zmq::message_t msgt;
    int nbytes;
    try{
      nbytes = sock.recv(&msgt, ZMQ_DONTWAIT);
    }catch(zmq::error_t e){
      return -1;
    }

    if(nbytes == 0){
      if(zmq_errno() == EAGAIN)
	return 0;
      else
	return -1;
    }

    size_t len = msgt.size();    
    uint8_t *dataptr;
    try{
      dataptr = new uint8_t[len];
    }catch(std::bad_alloc e){
      return -1;
    }
    
    memcpy(dataptr, msgt.data(), len);
    data.reset(dataptr);
    return len;
}

void HeaderManipulation::debugMsgBuffer(boost::shared_array<uint8_t> &msg_buffer, uint32_t &msg_size)
{
    for (uint j = 0; j < 4; ++j)
    {
        ROS_INFO("32-bit:%4i: %4i", j, (int)((uint32_t *)msg_buffer.get())[j]);
    }
    for (uint i = 0; i < 36; ++i)
    {
        ROS_DEBUG(" 8-bit:%4i: %2c = %4i", i, ((uint8_t *)msg_buffer.get())[i], (int)((uint8_t *)msg_buffer.get())[i]);
    }
    ROS_DEBUG("...");
    for (uint i = msg_size-6; i < msg_size; ++i)
    {
        ROS_DEBUG(" 8-bit:%4i: %2c = %4i", i, ((uint8_t *)msg_buffer.get())[i], (int)((uint8_t *)msg_buffer.get())[i]);
    }
}

void Header::write(const M_string& key_vals, boost::shared_array<uint8_t>& buffer, uint32_t& size)
{
    // Calculate the necessary size
    size = 0;
    {
        M_string::const_iterator it = key_vals.begin();
        M_string::const_iterator end = key_vals.end();
        for (; it != end; ++it)
        {
            const std::string& key = it->first;
            const std::string& value = it->second;

            size += key.length();
            size += value.length();
            size += 1; // = sign
            size += 4; // 4-byte length
        }
    }

    if (size == 0)
    {
        return;
    }

    buffer.reset(new uint8_t[size]);
    char* ptr = (char*)buffer.get();

    // Write the data
    {
        M_string::const_iterator it = key_vals.begin();
        M_string::const_iterator end = key_vals.end();
        for (; it != end; ++it)
        {
            const std::string& key = it->first;
            const std::string& value = it->second;

            uint32_t len = key.length() + value.length() + 1;
            SROS_SERIALIZE_PRIMITIVE(ptr, len);
            SROS_SERIALIZE_BUFFER(ptr, key.data(), key.length());
            static const char equals = '=';
            SROS_SERIALIZE_PRIMITIVE(ptr, equals);
            SROS_SERIALIZE_BUFFER(ptr, value.data(), value.length());
        }
    }

    ROS_ASSERT(ptr == (char*)buffer.get() + size);
}

static Future<string> _recv(
    const std::shared_ptr<SocketImpl>& impl,
    const Option<ssize_t>& size,
    Owned<string> buffer,
    size_t chunk,
    boost::shared_array<char> data,
    size_t length)
{
  if (length == 0) { // EOF.
    // Return everything we've received thus far, a subsequent receive
    // will return an empty string.
    return string(*buffer);
  }

  buffer->append(data.get(), length);

  if (size.isNone()) {
    // We've been asked just to return any data that we receive!
    return string(*buffer);
  } else if (size.get() < 0) {
    // We've been asked to receive until EOF so keep receiving since
    // according to the 'length == 0' check above we haven't reached
    // EOF yet.
    return impl->recv(data.get(), chunk)
      .then(lambda::bind(&_recv,
                         impl,
                         size,
                         buffer,
                         chunk,
                         data,
                         lambda::_1));
  } else if (static_cast<string::size_type>(size.get()) > buffer->size()) {
    // We've been asked to receive a particular amount of data and we
    // haven't yet received that much data so keep receiving.
    return impl->recv(data.get(), size.get() - buffer->size())
      .then(lambda::bind(&_recv,
                         impl,
                         size,
                         buffer,
                         chunk,
                         data,
                         lambda::_1));
  }

  // We've received as much data as requested, so return that data!
  return string(*buffer);
}

	void write_png_file(boost::shared_array<uint8_t> img, mem_encode* state)
	{
		/*
		static int test = 1;
		char filename[100];
		sprintf(filename, "test%d.png", test);
		++test;
		FILE* f = fopen(filename, "wb+");
		*/
		png_structp png_ptr;
		png_infop info_ptr;
		int number_of_passes;

		png_ptr = png_create_write_struct(PNG_LIBPNG_VER_STRING, NULL, NULL, NULL);

		png_bytep * row_pointers = (png_bytep*)png_malloc(png_ptr, 480 * sizeof(png_bytep));
		for(int h = 0; h < 480; ++h)
		{
			(row_pointers)[h] = (png_bytep)((char*)img.get()+(h*640*2));
		}
		/* initialize stuff */

		if (!png_ptr)
			abort_("[write_png_file] png_create_write_struct failed");

		/* if my_png_flush() is not needed, change the arg to NULL */
		png_set_write_fn(png_ptr, state, my_png_write_data, NULL);
		
		info_ptr = png_create_info_struct(png_ptr);
		if (!info_ptr)
			abort_("[write_png_file] png_create_info_struct failed");

		/* write header */
		if (setjmp(png_jmpbuf(png_ptr)))
			abort_("[write_png_file] Error during writing header");
	
		png_set_IHDR(png_ptr, info_ptr, 640, 480,
					 16, PNG_COLOR_TYPE_GRAY, PNG_INTERLACE_NONE,
					 PNG_COMPRESSION_TYPE_BASE, PNG_FILTER_TYPE_BASE);
	
		png_write_info(png_ptr, info_ptr);
		
		/* write bytes */
		if (setjmp(png_jmpbuf(png_ptr)))
			abort_("[write_png_file] Error during writing bytes");

		png_write_image(png_ptr, row_pointers);

		// /* end write */
		if (setjmp(png_jmpbuf(png_ptr)))
			abort_("[write_png_file] Error during end of write");

		png_write_end(png_ptr, NULL);

		/* cleanup heap allocation */
		png_free (png_ptr, row_pointers);
		png_destroy_write_struct (&png_ptr, &info_ptr);
		//fclose(f);
	}

PDU::PDU(const boost::shared_array<uint8_t> &pduLength, const boost::shared_array<uint8_t> &pduBuffer) :
		sb(""), buf(&sb), cmdId(0), cmdStatus(0), seqNo(0), nullTerminateOctetStrings(true), null(false)
{
	uint32_t bufSize = PDU::getPduLength(pduLength);

	buf.write(reinterpret_cast<char*>(pduLength.get()), HEADERFIELD_SIZE);
	if (buf.fail()) throw smpp::SmppException("PDU failed to write length");
	buf.write(reinterpret_cast<char*>(pduBuffer.get()), bufSize - HEADERFIELD_SIZE);
	if (buf.fail()) throw smpp::SmppException("PDU failed to write octets");

	buf.seekg(HEADERFIELD_SIZE, std::ios::cur);
	if (buf.fail()) throw smpp::SmppException("PDU failed to skip size header");

	(*this) >> cmdId;
	(*this) >> cmdStatus;
	(*this) >> seqNo;
}

void btlauncherAPI::gotDownloadProgram(const FB::JSObjectPtr& callback, 
									   std::string& program,
									   bool success,
									   const FB::HeaderMap& headers,
									   const boost::shared_array<uint8_t>& data,
									   const size_t size) {
	FBLOG_INFO("gotDownloadProgram()", "START");
	FBLOG_INFO("gotDownloadProgram()", program.c_str());
	

	char *tmpname = strdup("/tmp/btlauncherXXXXXX");
	mkstemp(tmpname);
	ofstream f(tmpname);
	
	if (f.fail()) {
		FBLOG_INFO("gotDownloadProgram()", "f.fail");
		callback->InvokeAsync("", FB::variant_list_of(false)(-1));
		return;
	}
	f.write((char *)data.get(), size);
	f.close();
	
	std::string url;
	if (program == "SoShare")
		url = SOSHARE_DOWNLOAD_URL;
	else if (program == "Torque")
		url = TORQUE_DOWNLOAD_URL;
	else
		url = BTLIVE_DOWNLOAD_URL;
	
	FBLOG_INFO("gotDownloadProgram()", url.c_str());
	
	const char *tarFlags = "-xf";
	if (url.find(".gz") != std::string::npos)
		tarFlags = "-xzf";
	
	pid_t tarPid;
	int status;
	switch(tarPid = fork()) 
	{
		case -1:
			FBLOG_INFO("gotDownloadProgram()", "fork failed");
			break;
		case 0:
			FBLOG_INFO("gotDownloadProgram()", "running tar");
			execl("/usr/bin/tar", "tar", tarFlags, tmpname, "-C", this->installPath.c_str(), NULL);
			break;
		default:
			FBLOG_INFO("gotDownloadProgram()", "waitpid");
			waitpid(tarPid, &status, 0);
			break;
	}
	
	runProgram(program, callback);
	FBLOG_INFO("gotDownloadProgram()", "END");
}

void kpoBaseApp::image_callback (const boost::shared_ptr<openni_wrapper::Image> &image)
{
    unsigned image_width_ = image->getWidth();
    unsigned image_height_ = image->getHeight();

    static unsigned rgb_array_size = 0;
    static boost::shared_array<unsigned char> rgb_array ((unsigned char*)(NULL));

    static unsigned char* rgb_buffer = 0;

    // here we need exact the size of the point cloud for a one-one correspondence!
    if (rgb_array_size < image_width_ * image_height_ * 3)
    {
      rgb_array_size = image_width_ * image_height_ * 3;
      rgb_array.reset (new unsigned char [rgb_array_size]);
      rgb_buffer = rgb_array.get ();
    }
    image->fillRGB (image_width_, image_height_, rgb_buffer, image_width_ * 3);

    {
        QMutexLocker locker (&mtx_);

        openniImage2opencvMat((XnRGB24Pixel*)rgb_buffer, scene_image_, image_height_, image_width_);

        if (need_image_cap) {
            std::cout << "need_image_cap " << need_image_cap << std::endl;
            cv::Mat rgb_image_;
            cv::cvtColor(scene_image_, rgb_image_, CV_BGR2RGB);

            qint64 timestamp = QDateTime::currentMSecsSinceEpoch();
            if (timestamp - last_snapshot_time > 30000) {
                last_snapshot_time = timestamp;
                QString filename = QString::fromUtf8("/myshare/autonomous_snapshots/");
                filename += QString::number(timestamp);
                filename += QString::fromUtf8(".png");

                cv::imwrite(filename.toStdString().c_str(), rgb_image_);
                need_image_cap = 0;
            }
        }
    }

}