void binary_reader::init(blob& bb)
 {
     _blob = bb;
     _size = bb.length();
     _ptr = bb.data();
     _remaining_size = _size;
 }

    int binary_reader::read(blob& blob)
    {
        int len;
        if (0 == read(len))
            return 0;

        if (len <= get_remaining_size())
        {
            blob = _blob.range(static_cast<int>(_ptr - _blob.data()), len);

            // optimization: zero-copy
            if (!blob.buffer_ptr())
            {
                std::shared_ptr<char> buffer(new char[len]);
                memcpy(buffer.get(), blob.data(), blob.length());
                blob = ::dsn::blob(buffer, 0, blob.length());
            }
            
            _ptr += len;
            _remaining_size -= len;
            return len + sizeof(len);
        }
        else
        {
            dassert(false, "read beyond the end of buffer");
            return 0;
        }
    }

 binary_reader::binary_reader(blob& blob)
 {
     _blob = blob;
     _size = blob.length();
     _ptr = blob.data();
     _remaining_size = _size;
 }

blob Base64::from(const blob& data) const {
	std::string transformed;
	CryptoPP::StringSource(data.data(), data.size(), true,
			new CryptoPP::Base64Decoder(
					new CryptoPP::StringSink(transformed)
			)
	);

	return blob(std::make_move_iterator(transformed.begin()), std::make_move_iterator(transformed.end()));
}

    binary_writer::binary_writer(blob& buffer)
    {
        _total_size = 0;
        _buffers.reserve(1);
        _reserved_size_per_buffer = _reserved_size_per_buffer_static;

        _buffers.push_back(buffer);
        _current_buffer = (char*)buffer.data();
        _current_offset = 0;
        _current_buffer_length = buffer.length();
    }

void Downloader::NeededChunk::put_block(uint32_t offset, const blob& content) {
	auto inserted = file_map_.insert({offset, content.size()}).second;
	if(inserted) {
#ifndef FOPEN_BACKEND
		std::copy(content.begin(), content.end(), mapped_file_.data()+offset);
#else
		if(wrapped_file_.ios().tellp() != offset)
			wrapped_file_.ios().seekp(offset);
		wrapped_file_.ios().write((char*)content.data(), content.size());
#endif
	}
}

 port& operator= (const port& rhs) {
     if (this != &rhs) {
         release(); m_blob = rhs.m_blob;
         if (m_blob) m_blob->inc_rc();
     }
     return *this;
 }

void P2PFolder::handle_message(const blob& message_raw) {
	V1Parser::message_type message_type = parser_.parse_MessageType(message_raw);

	counter_.add_down(message_raw.size());

	if(ready()) {
		switch(message_type) {
			case V1Parser::CHOKE: handle_Choke(message_raw); break;
			case V1Parser::UNCHOKE: handle_Unchoke(message_raw); break;
			case V1Parser::INTERESTED: handle_Interested(message_raw); break;
			case V1Parser::NOT_INTERESTED: handle_NotInterested(message_raw); break;
			case V1Parser::HAVE_META: handle_HaveMeta(message_raw); break;
			case V1Parser::HAVE_CHUNK: handle_HaveChunk(message_raw); break;
			case V1Parser::META_REQUEST: handle_MetaRequest(message_raw); break;
			case V1Parser::META_REPLY: handle_MetaReply(message_raw); break;
			case V1Parser::META_CANCEL: handle_MetaCancel(message_raw); break;
			case V1Parser::BLOCK_REQUEST: handle_BlockRequest(message_raw); break;
			case V1Parser::BLOCK_REPLY: handle_BlockReply(message_raw); break;
			case V1Parser::BLOCK_CANCEL: handle_BlockCancel(message_raw); break;
			default: throw protocol_error();
		}
	}else{
		handle_Handshake(message_raw);
	}
}

void meta_service::on_log_completed(error_code err, size_t size,
    blob buffer, 
    std::shared_ptr<configuration_update_request> req, dsn_message_t resp)
{
    dassert(err == ERR_OK, "log operation failed, cannot proceed, err = %s", err.to_string());
    dassert(buffer.length() == size, "log size must equal to the specified buffer size");

    configuration_update_response response;    
    update_configuration(*req, response);

    if (resp != nullptr)
    {
        meta_response_header rhdr;
        rhdr.err = err;
        rhdr.primary_address = primary_address();

        marshall(resp, rhdr);
        marshall(resp, response);

        dsn_rpc_reply(resp);
    }
    else
    {
        err.end_tracking();
    }
}

void Downloader::put_block(const blob& ct_hash, uint32_t offset, const blob& data, std::shared_ptr<RemoteFolder> from) {
	log_->trace() << log_tag() << BOOST_CURRENT_FUNCTION;
	auto needed_block_it = needed_chunks_.find(ct_hash);
	if(needed_block_it == needed_chunks_.end()) return;

	auto& requests = needed_block_it->second->requests;
	for(auto request_it = requests.begin(); request_it != requests.end();) {
		bool incremented_already = false;

		if(request_it->second.offset == offset          // Chunk position incorrect
			&& request_it->second.size == data.size()   // Chunk size incorrect
			&& request_it->first == from) {     // Requested node != replied. Well, it isn't critical, but will be useful to ban "fake" peers

			incremented_already = true;
			request_it = requests.erase(request_it);

			needed_block_it->second->put_block(offset, data);
			if(needed_block_it->second->full()) {
				exchange_group_.fs_dir()->put_chunk(ct_hash, needed_block_it->second->get_chunk());
			}   // TODO: catch "invalid hash" exception here

			periodic_maintain_.invoke_post();
		}

		if(!incremented_already) ++request_it;
	}
}

	inline reject_iter(dtable::iter * base, dtable * rejects, blob reject_value)
		: dtable_wrap_iter(base), rejects(rejects), reject_value(reject_value)
	{
		/* the reject value must exist, because nonexistent values
		 * can get pruned out if the shadow does not require them */
		assert(reject_value.exists());
	}

index_blob::index_blob(size_t count, const blob & x)
	: base(x), modified(false), resized(false), count(count)
{
	indices = new sub[count];
	if(!x.exists())
	{
		/* same as default constructor */
		modified = true;
		resized = true;
		return;
	}
	size_t offset = count * sizeof(uint32_t);
	for(size_t i = 0; i < count; i++)
	{
		uint32_t size = base.index<uint32_t>(i);
		if(size)
		{
			indices[i].delayed = true;
			indices[i]._size = --size;
			indices[i]._offset = offset;
			offset += size;
		}
		else
			assert(!indices[i].value.exists());
	}
}

 binary& operator= (const binary& rhs) {
     if (this != &rhs) {
         m_blob = rhs.m_blob;
         if (m_blob) m_blob->inc_rc();
     }
     return *this;
 }

 /**
  * Create a binary from the given data.
  * Data is shared between all cloned binaries by using reference counting.
  * @param data pointer to data.
  * @param size binary size in bytes
  * @param a_alloc is the allocator to use
  **/
 binary(const char* data, size_t size, const Alloc& a_alloc = Alloc()) {
     if (size == 0) {
         m_blob = nullptr;
         return;
     }
     m_blob = new blob<char, Alloc>(size, a_alloc);
     memcpy(m_blob->data(), data, size);
 }

void print(const blob & x, const char * prefix, ...)
{
	va_list ap;
	va_start(ap, prefix);
	if(!x.exists())
	{
		if(prefix)
			vprintf(prefix, ap);
		printf("(non-existent)\n");
		va_end(ap);
		return;
	}
	for(size_t i = 0; i < x.size(); i += 16)
	{
		size_t m = i + 16;
		if(prefix)
			vprintf(prefix, ap);
		for(size_t j = i; j < m; j++)
		{
			if(j < x.size())
				printf("%02x ", x[j]);
			else
				printf("   ");
			if((i % 16) == 8)
				printf(" ");
		}
		printf(" |");
		for(size_t j = i; j < m; j++)
		{
			if(j < x.size())
				printf("%c", isprint(x[j]) ? x[j] : '.');
			else
				printf(" ");
		}
		printf("|\n");
	}
	va_end(ap);
}

/* if we made a better find(), this could avoid flattening every time */
int simple_ctable::insert(const dtype & key, size_t column, const blob & value, bool append)
{
	int r = 0;
	assert(column < column_count);
	blob row = base->find(key);
	if(row.exists() || value.exists())
	{
		/* TODO: improve this... it is probably killing us */
		index_blob sub(column_count, row);
		sub.set(column, value);
		r = base->insert(key, sub.flatten(), append);
	}
	return r;
}

 port(const port& rhs) : m_blob(rhs.m_blob) { if (m_blob) m_blob->inc_rc(); }

 // Must only be called from constructor!
 void init(const atom& node, int id, uint8_t creation, 
           const Alloc& alloc) throw(err_bad_argument) 
 {
     m_blob = new blob<port_blob, Alloc>(1, alloc);
     new (m_blob->data()) port_blob(node, id & 0x0fffffff, creation & 0x03);
 }

 /**
  * Get the node name from the PORT.
  * @return the node name from the PORT.
  **/
 atom node() const { return m_blob ? m_blob->data()->node : atom::null; }

 void binary_writer::create_new_buffer(size_t size, /*out*/blob& bb)
 {
     std::shared_ptr<char> ptr(new char[size]);
     bb.assign(ptr, 0, (int)size);
 }