/*************************************************************************
*    UrBackup - Client/Server backup system
*    Copyright (C) 2011-2016 Martin Raiber
*
*    This program is free software: you can redistribute it and/or modify
*    it under the terms of the GNU Affero General Public License as published by
*    the Free Software Foundation, either version 3 of the License, or
*    (at your option) any later version.
*
*    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 Affero General Public License for more details.
*
*    You should have received a copy of the GNU Affero General Public License
*    along with this program.  If not, see <http://www.gnu.org/licenses/>.
**************************************************************************/

#include "../../Interface/Server.h"

#include "FileClient.h"

#include "../../fileservplugin/chunk_settings.h"

#include "../../common/data.h"
#include "../../stringtools.h"

#include "../../md5.h"
#include "../SparseFile.h"

#include <iostream>
#include <memory.h>
#include <algorithm>
#include <assert.h>
#include <limits.h>
#include <cstring>
#include <memory>

#ifndef _WIN32
#include <errno.h>
#endif

#ifdef __ANDROID__
extern "C" {
#include "ifaddrs.c"
}
#endif

namespace
{
	const std::string str_tmpdir="C:\\Windows\\Temp";
#ifndef _DEBUG
	const unsigned int DISCOVERY_TIMEOUT=30000; //30sec
#else
	const unsigned int DISCOVERY_TIMEOUT=1000; //1sec
#endif

	const size_t maxQueuedFiles = 3000;
	const size_t queuedFilesLow = 100;
	const char* multicast_group = "ff12::f894:d:dd00:ef91";
}

void Log(std::string str)
{
	Server->Log(str);
}

int curr_fnum=0;

bool setSockP(SOCKET sock)
{
#ifdef _WIN32
		DWORD dwBytesReturned = 0;
		BOOL bNewBehavior = FALSE;
		int status;

		// disable  new behavior using
		// IOCTL: SIO_UDP_CONNRESET
		#define SIO_UDP_CONNRESET _WSAIOW(IOC_VENDOR,12)
		status = WSAIoctl(sock, SIO_UDP_CONNRESET,
						&bNewBehavior, sizeof(bNewBehavior),
					NULL, 0, &dwBytesReturned,
					NULL, NULL);
		if (SOCKET_ERROR == status)
		{
			return false;
		}
#endif
        return true;
}    

FileClient::FileClient(bool enable_find_servers, std::string identity, int protocol_version, bool add_request_checksums,
	FileClient::ReconnectionCallback *reconnection_callback, FileClient::NoFreeSpaceCallback *nofreespace_callback)
	: tcpsock(NULL), starttime(0), connect_starttime(0), socket_open(false), connected(false),
	local_ip(),
	max_version(), server_addr(), connection_id(), 
	protocol_version(protocol_version), 
	transferred_bytes(0), reconnection_callback(reconnection_callback),
	nofreespace_callback(nofreespace_callback), reconnection_timeout(300000), retryBindToNewInterfaces(true),
	identity(identity), received_data_bytes(0), queue_callback(NULL), dl_off(0),
	last_transferred_bytes(0), last_progress_log(0), progress_log_callback(NULL), needs_flush(false),
	real_transferred_bytes(0), is_downloading(false), sparse_extends_f(NULL), sparse_bytes(0),
	reconnect_tries(50)
{
	memset(buffer, 0, BUFFERSIZE_UDP);

	if(enable_find_servers)
	{
		bindToNewInterfaces();
	}

	socket_open=false;
	stack.setAddChecksum(add_request_checksums);

	mutex = Server->createMutex();
}

int FileClient::getReconnectTriesDecr()
{
	IScopedLock lock(mutex);
	return reconnect_tries--;
}

void FileClient::bindToNewInterfaces()
{
	std::string s_broadcast_source_port = Server->getServerParameter("broadcast_source_port");
	unsigned short broadcast_source_port = UDP_SOURCE_PORT;
	if (!s_broadcast_source_port.empty())
	{
		broadcast_source_port = static_cast<unsigned short>(atoi(s_broadcast_source_port.c_str()));
	}

#ifndef _WIN32
	std::string bcast_interfaces = Server->getServerParameter("broadcast_interfaces", "");

	std::vector<std::string> bcast_filter;
	if (!bcast_interfaces.empty())
	{
		Tokenize(bcast_interfaces, bcast_filter, ";,");
	}

	ifaddrs *start_ifap;
	int rc = getifaddrs(&start_ifap);
	if (rc == 0)
	{
		ifaddrs* ifap = start_ifap;
		for (; ifap != NULL; ifap = ifap->ifa_next)
		{
			bool found_name = bcast_filter.empty() || std::find(bcast_filter.begin(), bcast_filter.end(), ifap->ifa_name) != bcast_filter.end();

			if (found_name &&
				!(ifap->ifa_flags & IFF_LOOPBACK)
				&& !(ifap->ifa_flags & IFF_POINTOPOINT))
			{
				if (ifap->ifa_addr!=NULL
					&& ifap->ifa_addr->sa_family == AF_INET
					&& (ifap->ifa_flags & IFF_BROADCAST))
				{
					sockaddr_in source_addr;
					memset(&source_addr, 0, sizeof(source_addr));
					source_addr.sin_addr = ((struct sockaddr_in *)ifap->ifa_addr)->sin_addr;
					source_addr.sin_family = AF_INET;
					source_addr.sin_port = htons(broadcast_source_port);


					if (alreadyHasAddrv4(source_addr))
						continue;

					int type = SOCK_DGRAM;
#if defined(SOCK_CLOEXEC)
					type |= SOCK_CLOEXEC;
#endif
					SOCKET udpsock = socket(AF_INET, type, 0);
					if (udpsock == -1)
					{
						Server->Log(std::string("Error creating socket for interface ") + std::string(ifap->ifa_name), LL_ERROR);
						continue;
					}
#if !defined(SOCK_CLOEXEC)
					fcntl(udpsock, F_SETFD, fcntl(udpsock, F_GETFD, 0) | FD_CLOEXEC);
#endif

					BOOL val = TRUE;
					int rc = setsockopt(udpsock, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(BOOL));
					if (rc < 0)
					{
						Server->Log(std::string("Setting SO_REUSEADDR failed for interface ") + std::string(ifap->ifa_name), LL_ERROR);
					}

					Server->Log(std::string("Binding to interface ") + std::string(ifap->ifa_name) + " (ipv4) for broadcasting...", LL_DEBUG);

					rc = bind(udpsock, (struct sockaddr *)&source_addr, sizeof(source_addr));
					if (rc < 0)
					{
						Server->Log(std::string("Binding UDP socket failed for interface ") + std::string(ifap->ifa_name), LL_ERROR);
					}

					rc = setsockopt(udpsock, SOL_SOCKET, SO_BROADCAST, (char*)&val, sizeof(BOOL));
					if (rc < 0)
					{
						Server->Log(std::string("Enabling SO_BROADCAST for UDP socket failed for interface ") + std::string(ifap->ifa_name), LL_ERROR);
						closesocket(udpsock);
						continue;
					}

#if defined(__FreeBSD__)
					int optval = 1;
					if (setsockopt(udpsock, IPPROTO_IP, IP_ONESBCAST, &optval, sizeof(int)) == -1)
					{
						Server->Log(std::string("Error setting IP_ONESBCAST for interface ") + std::string(ifap->ifa_name), LL_ERROR);
					}
#endif


					SSocket new_udpsock;

					new_udpsock.is_ipv6 = false;
					new_udpsock.addr_ipv4 = source_addr;
					new_udpsock.udpsock = udpsock;
					new_udpsock.broadcast_addr = *((struct sockaddr_in *)ifap->ifa_broadaddr);

					udpsocks.push_back(new_udpsock);

					Server->Log("Broadcasting on ipv4 interface "+std::string(ifap->ifa_name) + " addr " + ipToString(new_udpsock));
				}
				else if (ifap->ifa_addr != NULL 
					&& ifap->ifa_addr->sa_family == AF_INET6
					&& Server->getServerParameter("disable_ipv6").empty())
				{
					sockaddr_in6 source_addr;
					memset(&source_addr, 0, sizeof(source_addr));
					source_addr.sin6_addr = ((struct sockaddr_in6 *)ifap->ifa_addr)->sin6_addr;
					source_addr.sin6_family = AF_INET6;
					source_addr.sin6_port = htons(broadcast_source_port);
					source_addr.sin6_scope_id = ((struct sockaddr_in6 *)ifap->ifa_addr)->sin6_scope_id;


					if (alreadyHasAddrv6(source_addr))
						continue;

					int type = SOCK_DGRAM;
#if defined(SOCK_CLOEXEC)
					type |= SOCK_CLOEXEC;
#endif
					SOCKET udpsock = socket(AF_INET6, type, 0);
					if (udpsock == -1)
					{
						Server->Log(std::string("Error creating ipv6 socket for interface ") + std::string(ifap->ifa_name), LL_ERROR);
						continue;
					}
#if !defined(SOCK_CLOEXEC)
					fcntl(udpsock, F_SETFD, fcntl(udpsock, F_GETFD, 0) | FD_CLOEXEC);
#endif

					BOOL val = TRUE;
					int rc = setsockopt(udpsock, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(BOOL));
					if (rc < 0)
					{
						Server->Log(std::string("Setting SO_REUSEADDR failed for interface ") + std::string(ifap->ifa_name), LL_ERROR);
					}

					int optval = 1;
					setsockopt(udpsock, IPPROTO_IPV6, IPV6_V6ONLY, (char*)&optval, sizeof(optval));

					Server->Log(std::string("Binding to interface ") + std::string(ifap->ifa_name) + " (ipv6) for broadcasting...", LL_DEBUG);

					SSocket new_udpsock;

					new_udpsock.is_ipv6 = true;
					new_udpsock.addr_ipv6 = source_addr;
					new_udpsock.udpsock = udpsock;

					rc = bind(udpsock, (struct sockaddr *)&source_addr, sizeof(source_addr));
					if (rc < 0)
					{
						Server->Log(std::string("Binding UDP socket failed for interface ") + std::string(ifap->ifa_name)+" errno="+convert(errno)+" interface ipv6 "+ipToString(new_udpsock), LL_ERROR);
					}


					udpsocks.push_back(new_udpsock);

					Server->Log("Broadcasting on ipv6 interface "+ std::string(ifap->ifa_name) +" addr " + ipToString(new_udpsock));
				}
			}
		}
		freeifaddrs(start_ifap);
	}
	else
	{
		retryBindToNewInterfaces = false;

		Server->Log("Getting interface ips failed. errno=" + convert(errno) +
			". Server may not listen properly on all network devices when discovering clients.", LL_ERROR);

		int type = SOCK_DGRAM;
#if defined(SOCK_CLOEXEC)
		type |= SOCK_CLOEXEC;
#endif
		SOCKET udpsock = socket(AF_INET, type, 0);
		if (udpsock == -1)
		{
			Server->Log("Error creating socket", LL_ERROR);
		}
		else
		{
#if !defined(SOCK_CLOEXEC)
			fcntl(udpsock, F_SETFD, fcntl(udpsock, F_GETFD, 0) | FD_CLOEXEC);
#endif
			BOOL val = TRUE;
			int rc = setsockopt(udpsock, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(BOOL));
			if (rc < 0)
			{
				Server->Log("Setting SO_REUSEADDR failed", LL_ERROR);
			}

			sockaddr_in source_addr;
			memset(&source_addr, 0, sizeof(source_addr));
			source_addr.sin_addr.s_addr = htonl(INADDR_ANY);
			source_addr.sin_family = AF_INET;
			source_addr.sin_port = htons(broadcast_source_port);

			Server->Log("Binding to no interface for broadcasting. Entering IP on restore CD won't work.", LL_WARNING);

			rc = setsockopt(udpsock, SOL_SOCKET, SO_BROADCAST, (char*)&val, sizeof(BOOL));
			if (rc < 0)
			{
				Server->Log("Enabling SO_BROADCAST for UDP socket failed", LL_ERROR);
				closesocket(udpsock);
			}
			else
			{
				SSocket new_udpsock;

				source_addr.sin_addr.s_addr = INADDR_BROADCAST;

				new_udpsock.is_ipv6 = false;
				new_udpsock.addr_ipv4 = source_addr;
				new_udpsock.udpsock = udpsock;
				new_udpsock.broadcast_addr = source_addr;

				udpsocks.push_back(new_udpsock);

				Server->Log("Broadcasting on interface IPv4 " + ipToString(new_udpsock));
			}
		}

		if (Server->getServerParameter("disable_ipv6").empty())
		{
			SOCKET udpsock = socket(AF_INET6, type, 0);
			if (udpsock == -1)
			{
				Server->Log("Error creating ipv6 socket", LL_ERROR);
			}
			else
			{
#if !defined(SOCK_CLOEXEC)
				fcntl(udpsock, F_SETFD, fcntl(udpsock, F_GETFD, 0) | FD_CLOEXEC);
#endif
				BOOL val = TRUE;
				int rc = setsockopt(udpsock, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(BOOL));
				if (rc < 0)
				{
					Server->Log("Setting SO_REUSEADDR failed", LL_ERROR);
				}

				int optval = 1;
				setsockopt(udpsock, IPPROTO_IPV6, IPV6_V6ONLY, (char*)&optval, sizeof(optval));

				sockaddr_in6 source_addr;
				memset(&source_addr, 0, sizeof(source_addr));
				source_addr.sin6_addr = in6addr_any;
				source_addr.sin6_family = AF_INET6;
				source_addr.sin6_port = htons(broadcast_source_port);

				Server->Log("Binding to no interface for broadcasting. Entering IP on restore CD won't work.", LL_WARNING);

				SSocket new_udpsock;
				new_udpsock.is_ipv6 = true;
				new_udpsock.addr_ipv6 = source_addr;
				new_udpsock.udpsock = udpsock;

				udpsocks.push_back(new_udpsock);

				Server->Log("Broadcasting on interface IPv6 " + ipToString(new_udpsock));
			}
		}
	}
#else
	char hostname[MAX_PATH];

    _i32 rc=gethostname(hostname, MAX_PATH);
    if(rc==SOCKET_ERROR)
	{
		Server->Log("Error getting Hostname", LL_ERROR);
		retryBindToNewInterfaces=false;
        return;
	}
	std::vector<_u32> addresses;

	ADDRINFOA hints = {};
	hints.ai_flags = AI_PASSIVE;
	hints.ai_family = AF_INET;

	PADDRINFOA h = NULL;
	rc = getaddrinfo(hostname, NULL, &hints, &h);
	PADDRINFOA orig_h = h;

    if(rc==0 && h!=NULL)
    {
		if(h->ai_family!=AF_INET)
		{
			Server->Log("Hostname hostent is not AF_INET (ipv4)", LL_ERROR);
		}
		else
		{
			while(h!=NULL)
			{
				sockaddr_in source_addr;
				memset(&source_addr, 0, sizeof(source_addr));
				source_addr.sin_family = AF_INET;
				source_addr.sin_addr = reinterpret_cast<sockaddr_in*>(h->ai_addr)->sin_addr;
				source_addr.sin_port = htons(broadcast_source_port);

				if (alreadyHasAddrv4(source_addr))
				{
					h = h->ai_next;
					continue;
				}

				SOCKET udpsock=socket(AF_INET,SOCK_DGRAM,0);	

				int optval=1;
				int rc=setsockopt(udpsock, SOL_SOCKET, SO_REUSEADDR, (char*)&optval, sizeof(int));
				if(rc==SOCKET_ERROR)
				{
					Server->Log("Failed setting SO_REUSEADDR in FileClient", LL_ERROR);
				}

				rc = bind(udpsock, (struct sockaddr *)&source_addr, sizeof(source_addr));
				if(rc<0)
				{
					Server->Log("Binding UDP socket failed", LL_ERROR);
				}

				setSockP(udpsock);

				BOOL val=TRUE;
				rc=setsockopt(udpsock, SOL_SOCKET, SO_BROADCAST, (char*)&val, sizeof(BOOL) );      
				if(rc<0)
				{
					Server->Log("Failed setting SO_BROADCAST in FileClient", LL_ERROR);
				}

				SSocket new_udpsock;

				new_udpsock.is_ipv6 = false;
				new_udpsock.addr_ipv4 = source_addr;
				new_udpsock.udpsock = udpsock;

				udpsocks.push_back(new_udpsock);

				Server->Log("Broadcasting on interface IPv4 "+ ipToString(new_udpsock));

				h = h->ai_next;
			}
		}				
    }

	if (orig_h != NULL)
		freeaddrinfo(orig_h);


	if (Server->getServerParameter("disable_ipv6").empty())
	{
		hints.ai_family = AF_INET6;
		PADDRINFOA h = NULL;
		rc = getaddrinfo(hostname, NULL, &hints, &h);
		PADDRINFOA orig_h = h;

		if (rc == 0 && h != NULL)
		{
			if (h->ai_family != AF_INET6)
			{
				Server->Log("Hostname hostent is not AF_INET6 (ipv6)", LL_ERROR);
			}
			else
			{
				while (h != NULL)
				{
					sockaddr_in6 source_addr;
					memset(&source_addr, 0, sizeof(source_addr));
					source_addr.sin6_family = AF_INET6;
					source_addr.sin6_addr = reinterpret_cast<sockaddr_in6*>(h->ai_addr)->sin6_addr;
					source_addr.sin6_port = htons(broadcast_source_port);

					if (alreadyHasAddrv6(source_addr))
					{
						h = h->ai_next;
						continue;
					}

					SOCKET udpsock = socket(AF_INET6, SOCK_DGRAM, 0);

					int optval = 1;
					int rc = setsockopt(udpsock, SOL_SOCKET, SO_REUSEADDR, (char*)&optval, sizeof(int));
					if (rc == SOCKET_ERROR)
					{
						Server->Log("Failed setting SO_REUSEADDR in FileClient (2)", LL_ERROR);
					}

					optval = 1;
					setsockopt(udpsock, IPPROTO_IPV6, IPV6_V6ONLY, (char*)&optval, sizeof(optval));

					rc = bind(udpsock, (struct sockaddr *)&source_addr, sizeof(source_addr));
					if (rc<0)
					{
						Server->Log("Binding UDP socket failed (2)", LL_ERROR);
					}

					setSockP(udpsock);

					SSocket new_udpsock;

					new_udpsock.is_ipv6 = true;
					new_udpsock.addr_ipv6 = source_addr;
					new_udpsock.udpsock = udpsock;

					udpsocks.push_back(new_udpsock);

					Server->Log("Broadcasting on interface IPv6 " + ipToString(new_udpsock));

					h = h->ai_next;
				}
			}
		}

		if (orig_h != NULL)
			freeaddrinfo(orig_h);
	}

#endif
}

FileClient::~FileClient(void)
{
	if(socket_open && tcpsock!=NULL)
	{
		Server->destroy(tcpsock);
	}
	for(size_t i=0;i<udpsocks.size();++i)
	{
		closesocket(udpsocks[i].udpsock);
	}

	Server->destroy(mutex);

	if (sparse_extends_f != NULL)
	{
		std::string tmpfn = sparse_extends_f->getFilename();
		Server->destroy(sparse_extends_f);
		Server->deleteFile(tmpfn);
	}
}

std::vector<FileClient::SAddrHint> FileClient::getServers(void)
{
        return servers;
}

std::vector<std::string> FileClient::getServerNames(void)
{
        return servernames;
}

std::vector<FileClient::SAddrHint> FileClient::getWrongVersionServers(void)
{
        return wvservers;
}

_u32 FileClient::getLocalIP(void)
{
        return local_ip;
}

_u32 FileClient::GetServers(bool start, const std::vector<SAddrHint> &addr_hints)
{
        if(start)
        {
			if(retryBindToNewInterfaces)
			{
				bindToNewInterfaces();
			}

			max_version=0;
			for(size_t i=0;i<udpsocks.size();++i)
			{
				if (!udpsocks[i].is_ipv6)
				{
					sockaddr_in addr_udp;
					addr_udp.sin_family = AF_INET;
					addr_udp.sin_port = htons(UDP_PORT);
#ifdef __FreeBSD__
					addr_udp.sin_addr.s_addr = udpsocks[i].broadcast_addr.sin_addr.s_addr;
#else
					addr_udp.sin_addr.s_addr = INADDR_BROADCAST;
#endif
					memset(addr_udp.sin_zero, 0, sizeof(addr_udp.sin_zero));

					char ch = ID_PING;
					int rc = sendto(udpsocks[i].udpsock, &ch, 1, 0, (sockaddr*)&addr_udp, sizeof(addr_udp));
					if (rc == -1)
					{
						Server->Log("Sending broadcast failed!", LL_ERROR);
					}
				}
				else
				{
					sockaddr_in6 addr_udp = {};
					addr_udp.sin6_family = AF_INET6;
					addr_udp.sin6_port = htons(UDP_PORT);
					if (inet_pton(AF_INET6, multicast_group, &addr_udp.sin6_addr) != 1)
					{
						Server->Log("inet_pton failed", LL_ERROR);
					}

					char ch = ID_PING;
					int rc = sendto(udpsocks[i].udpsock, &ch, 1, 0, (sockaddr*)&addr_udp, sizeof(addr_udp));
					if (rc == -1)
					{
						Server->Log("Sending broadcast failed! (ipv6)", LL_ERROR);
					}
				}
			}

			if(!addr_hints.empty())
			{
				for(size_t i=0;i<udpsocks.size();++i)
				{
					int broadcast=0;
#ifdef _WIN32
#define SETSOCK_CAST (char*)
#else
#define SETSOCK_CAST
#endif

					if(!udpsocks[i].is_ipv6)
					{
						if(setsockopt(udpsocks[i].udpsock, SOL_SOCKET, SO_BROADCAST, SETSOCK_CAST &broadcast, sizeof(int))==-1)
						{
							Server->Log("Error setting socket to not broadcast", LL_ERROR);
						}

						#if defined(__FreeBSD__)
						int optval=0;
						if(setsockopt(udpsocks[i].udpsock, IPPROTO_IP, IP_ONESBCAST, &optval, sizeof(int))==-1)
						{
							Server->Log(std::string("Error setting IP_ONESBCAST" ), LL_ERROR);
						}
						#endif

					}

					for(size_t j=0;j<addr_hints.size();++j)
					{
						char ch=ID_PING;

						if (!addr_hints[j].is_ipv6)
						{
							sockaddr_in addr_udp = {};
							addr_udp.sin_family = AF_INET;
							addr_udp.sin_port = htons(UDP_PORT);
							addr_udp.sin_addr.s_addr = addr_hints[j].addr_ipv4;

							sendto(udpsocks[i].udpsock, &ch, 1, 0, (sockaddr*)&addr_udp, sizeof(sockaddr_in));
						}
						else
						{
							sockaddr_in6 addr_udp = {};
							addr_udp.sin6_family = AF_INET6;
							addr_udp.sin6_port = htons(UDP_PORT);
							memcpy(&addr_udp.sin6_addr, addr_hints[j].addr_ipv6, sizeof(addr_hints[j].addr_ipv6));

							sendto(udpsocks[i].udpsock, &ch, 1, 0, (sockaddr*)&addr_udp, sizeof(sockaddr_in));
						}
					}

					if (!udpsocks[i].is_ipv6)
					{
						broadcast = 1;
						if (setsockopt(udpsocks[i].udpsock, SOL_SOCKET, SO_BROADCAST, SETSOCK_CAST &broadcast, sizeof(int)) == -1)
						{
							Server->Log("Error setting socket to broadcast", LL_ERROR);
						}

#undef SETSOCK_CAST

#if defined(__FreeBSD__)
						int optval = 1;
						if (setsockopt(udpsocks[i].udpsock, IPPROTO_IP, IP_ONESBCAST, &optval, sizeof(int)) == -1)
						{
							Server->Log(std::string("Error setting IP_ONESBCAST"), LL_ERROR);
						}
#endif
					}
				}
			}

			starttime=Server->getTimeMS();

			servers.clear();
			servernames.clear();
			wvservers.clear(); 


			return ERR_CONTINUE;
        }
        else
        {
#ifdef _WIN32
			fd_set fdset;
			FD_ZERO(&fdset);

			SOCKET max_socket;

			if(!udpsocks.empty())
			{
				max_socket=udpsocks[0].udpsock;
			}

			for(size_t i=0;i<udpsocks.size();++i)
			{
				FD_SET(udpsocks[i].udpsock, &fdset);
				max_socket=(std::max)(max_socket, udpsocks[i].udpsock);
			}

			timeval lon;
			lon.tv_sec=0;
			lon.tv_usec=1000*1000;
			_i32 rc = select((int)max_socket+1, &fdset, 0, 0, &lon);
#else
			std::vector<pollfd> conn;
			conn.resize(udpsocks.size());
			for(size_t i=0;i<udpsocks.size();++i)
			{
				conn[i].fd=udpsocks[i].udpsock;
				conn[i].events=POLLIN;
				conn[i].revents=0;
			}
			int rc = poll(&conn[0], conn.size(), 1000);
#endif
        	if(rc>0)
	        {

				for(size_t i=0;i<udpsocks.size();++i)
				{
#ifdef _WIN32
					if(FD_ISSET(udpsocks[i].udpsock, &fdset))
#else
					if(conn[i].revents!=0)
#endif
					{
						socklen_t addrsize;
						sockaddr* sender;
						sockaddr_in6 sender_v6;
						sockaddr_in sender_v4;
						if (udpsocks[i].is_ipv6)
						{
							sender = (sockaddr*)&sender_v6;
							addrsize = sizeof(sender_v6);
						}
						else
						{
							sender = (sockaddr*)&sender_v4;
							addrsize = sizeof(sender_v4);
						}
						_i32 err = recvfrom(udpsocks[i].udpsock, buffer, BUFFERSIZE_UDP, 0, sender, &addrsize);
						if(err==SOCKET_ERROR)
						{
							continue;
        				}
						if(err>2&&buffer[0]==ID_PONG)
						{
							int version=(unsigned char)buffer[1];
							if(version==FILECLIENT_VERSION)
							{
								std::string sn;

								sn.resize(err - 2);
								memcpy((char*)sn.c_str(), &buffer[2], err - 2);

								if (!sn.empty())
								{
									if (udpsocks[i].is_ipv6)
									{
										SAddrHint res;
										res.is_ipv6 = true;
										memcpy(res.addr_ipv6, &sender_v6.sin6_addr, sizeof(sender_v6.sin6_addr));
										res.zone = sender_v6.sin6_scope_id;
										servers.push_back(res);
									}
									else
									{
										SAddrHint res;
										res.is_ipv6 = false;
										res.addr_ipv4 = sender_v4.sin_addr.s_addr;
										servers.push_back(res);
									}									
									servernames.push_back(sn);
								}
							}
							else
							{
								if (udpsocks[i].is_ipv6)
								{
									SAddrHint res;
									res.is_ipv6 = true;
									memcpy(res.addr_ipv6, &sender_v6.sin6_addr, sizeof(sender_v6.sin6_addr));
									res.zone = sender_v6.sin6_scope_id;
									wvservers.push_back(res);
								}
								else
								{
									SAddrHint res;
									res.is_ipv6 = false;
									res.addr_ipv4 = sender_v4.sin_addr.s_addr;
									wvservers.push_back(res);
								}
							}
                                
							if( version>max_version )
							{
								max_version=version;
							}
						}
					}
				}
			}


            if(Server->getTimeMS()-starttime>DISCOVERY_TIMEOUT)
            {
                    return ERR_TIMEOUT;
            }
            else
                    return ERR_CONTINUE;

        }

        
}

int FileClient::getMaxVersion(void)
{
        return max_version;        
}

_u32 FileClient::Connect(const SAddrHint& addr)
{
	if( socket_open==true )
    {
            Server->destroy(tcpsock);
    }

	tcpsock=Server->ConnectStream(addr.toString(), TCP_PORT, 10000);

	if(tcpsock!=NULL)
	{
		socket_open=true;

		for(size_t i=0;i<throttlers.size();++i)
		{
			tcpsock->addThrottler(throttlers[i]);
		}
	}

	server_addr=addr;

    if(tcpsock==NULL)
		return ERR_ERROR;
	else
		return ERR_CONNECTED;
}    

void FileClient::addThrottler(IPipeThrottler *throttler)
{
	throttlers.push_back(throttler);
	if(tcpsock!=NULL)
	{
		tcpsock->addThrottler(throttler);
	}
}

_u32 FileClient::Connect(IPipe *cp)
{
	if( socket_open==true )
    {
            Server->destroy(tcpsock);
    }

	tcpsock=cp;

	if(tcpsock!=NULL)
	{
		socket_open=true;
	}

	if(tcpsock==NULL)
		return ERR_ERROR;
	else
		return ERR_CONNECTED;
}

void FileClient::setServerName(std::string pName)
{
        mServerName=pName;
}

std::string FileClient::getServerName(void)
{
        return mServerName;
}

bool FileClient::isConnected(void)
{
        return connected;
}

bool FileClient::Reconnect(void)
{
	dl_off=0;
	queued.clear();
	if(queue_callback!=NULL)
	{
		queue_callback->resetQueueFull();
	}

	unsigned int local_reconnection_timeout;
	{
		IScopedLock lock(mutex);
		if (tcpsock != NULL)
		{
			IScopedLock lock(mutex);
			transferred_bytes += tcpsock->getTransferedBytes();
			real_transferred_bytes += tcpsock->getRealTransferredBytes();
			Server->destroy(tcpsock);
			tcpsock = NULL;
		}

		local_reconnection_timeout = reconnection_timeout;
	}
	connect_starttime=Server->getTimeMS();
	
	while(Server->getTimeMS()-connect_starttime<local_reconnection_timeout)
	{
		IPipe* new_tcpsock;
		if(reconnection_callback==NULL)
		{
			new_tcpsock=Server->ConnectStream(server_addr.toString(), TCP_PORT, 10000);
		}
		else
		{
			new_tcpsock=reconnection_callback->new_fileclient_connection();
		}
		if(new_tcpsock!=NULL)
		{
			for(size_t i=0;i<throttlers.size();++i)
			{
				new_tcpsock->addThrottler(throttlers[i]);
			}
			Server->Log("Reconnected successfully,", LL_DEBUG);
			IScopedLock lock(mutex);
			tcpsock = new_tcpsock;
			socket_open=true;
			return true;
		}
		else
		{
			Server->wait(1000);
			IScopedLock lock(mutex);
			local_reconnection_timeout = reconnection_timeout;
		}
	}
	Server->Log("Reconnecting failed.", LL_DEBUG);
	socket_open=false;
	return false;
}

 _u32 FileClient::GetFile(std::string remotefn, IFsFile *backing_file, bool hashed, bool metadata_only, size_t folder_items, bool is_script, size_t file_id)
{
	if(tcpsock==NULL)
		return ERR_ERROR;

	resetSparseExtentsFile();

	{
		IScopedLock lock(mutex);
		if (reconnect_tries < 50)
			reconnect_tries = 50;
	}

	if(!hashed && protocol_version>1)
	{
		//Disable hashed transfer (protocol_version>1)
		protocol_version=1;
	}

	if(queued.empty())
	{
		CWData data;
		data.addUChar( metadata_only? ID_GET_FILE_METADATA_ONLY : (
				file_id!=0 ? ID_GET_FILE_WITH_METADATA :
				(protocol_version>1?ID_GET_FILE_RESUME_HASH:ID_GET_FILE_RESUME) ) );
		data.addString( remotefn );
		data.addString( identity );

		if(metadata_only)
		{
			data.addChar(0);
			data.addVarInt(folder_items);
			data.addVarInt(file_id);
		}
		else if(file_id!=0)
		{
			data.addChar(0);
			data.addChar(hashed?1:0);
			data.addVarInt(file_id);
			data.addChar(1);
		}

		needs_flush = true;

		while(stack.Send( tcpsock, data.getDataPtr(), data.getDataSize() )!=data.getDataSize())
		{
			Server->Log("Timeout during file request (1). Reconnecting...", LL_DEBUG);

			bool b = false;
			int tries = getReconnectTriesDecr();
			if (tries > 0)
			{
				b = Reconnect();
			}
			if (!b)
			{
				Server->Log("Timeout during file request (1)", LL_ERROR);
				return ERR_TIMEOUT;
			}
		}
	}
	else
	{
		assert(queued.front().fn == remotefn);
		assert(!queued.front().finish_script);
		queued.pop_front();
	}

	_u64 filesize=0;
	_u64 received=0;
	_u64 next_checkpoint=c_checkpoint_dist;
	_u64 last_checkpoint=0;
	_u64 num_sparse_extends = 0;
	bool firstpacket=true;
	last_progress_log=0;

    starttime=Server->getTimeMS();
	int64 last_timeouttime = 0;


	enum EReceiveState
	{
		EReceiveState_Data=0,
		EReceiveState_Hash=1,
		EReceiveState_SparseExtends=2
	};

	EReceiveState state = EReceiveState_Data;
	char hash_buf[16];
	_u32 hash_r;
	MD5 hash_func;
	IFile* file = backing_file;
	std::unique_ptr<SparseFile> sparse_file;

	char* buf = dl_buf;

	_u32 queue_err = ERR_SUCCESS;

	while(true)
	{        
		if(!is_script)
		{
			queue_err = fillQueue();
		}
		else
		{
			if(needs_flush)
			{
				needs_flush=false;
				Flush();
			}
		}

		size_t rc;
		if (queue_err != ERR_SUCCESS)
		{
			Server->Log("Queue err " + convert(queue_err), LL_DEBUG);
			rc = 0;
		}
		else if(tcpsock->isReadable() || dl_off==0 ||
			(firstpacket 
				&& ( (dl_buf[0]==ID_FILESIZE && dl_off<1+sizeof(_u64))
					|| (dl_buf[0] == ID_FILESIZE_AND_EXTENTS && dl_off<1 + 2*sizeof(_u64)) )
			) )
		{
			rc = tcpsock->Read(&dl_buf[dl_off], BUFFERSIZE-dl_off, 120000);

			if (rc != 0)
			{
				rc += dl_off;
			}
		}
		else
		{
			rc = dl_off;
		}
		dl_off=0;

        if( rc==0 )
        {
			Server->Log("Server timeout (2) in FileClient", LL_DEBUG);

			if (last_timeouttime == 0
				|| Server->getTimeMS() - last_timeouttime>30*60*1000)
			{
				IScopedLock lock(mutex);
				if (reconnect_tries < 50)
					reconnect_tries = 50;
			}

			last_timeouttime = Server->getTimeMS();

			bool b = false;
			int tries = getReconnectTriesDecr();
			if (tries > 0)
			{
				b = Reconnect();
			}
			else
			{
				Server->Log("FileClient: Reconnect tries exhausted (1)", LL_INFO);
			}
			if(!b )
			{
				Server->Log("FileClient: ERR_TIMEOUT", LL_INFO);
				return ERR_TIMEOUT;
			}
			else
			{
				CWData data;
				data.addUChar( metadata_only?ID_GET_FILE_METADATA_ONLY : 
					( file_id!=0 ? ID_GET_FILE_WITH_METADATA :
					(protocol_version>1?ID_GET_FILE_RESUME_HASH:ID_GET_FILE_RESUME) ) );
				data.addString( remotefn );
				data.addString( identity );

				if(metadata_only)
				{
					data.addChar(0);
					data.addVarInt(folder_items);
					data.addVarInt(file_id);
				}
				else if(file_id!=0)
				{
					data.addChar(0);
					data.addChar(hashed?1:0);
					data.addVarInt(file_id);
					data.addChar(1);
				}

				bool resume = received > 0;

				if( protocol_version>1 )
				{
					received=last_checkpoint;
				}

				if (resume || received>0)
				{
					data.addInt64(received);
				}

				if(file!=NULL)
				{
					file->Seek(received);
				}

				while (stack.Send(tcpsock, data.getDataPtr(), data.getDataSize()) != data.getDataSize())
				{
					Server->Log("Timeout during file request (4). Reconnecting...", LL_DEBUG);

					if (last_timeouttime == 0
						|| Server->getTimeMS() - last_timeouttime>30 * 60 * 1000)
					{
						IScopedLock lock(mutex);
						if (reconnect_tries < 50)
							reconnect_tries = 50;
					}

					last_timeouttime = Server->getTimeMS();

					bool b = false;
					int tries = getReconnectTriesDecr();
					if (tries > 0)
					{
						b = Reconnect();
					}
					else
					{
						Server->Log("FileClient: Reconnect tries exhausted (2)", LL_INFO);
					}
					if (!b)
					{
						Server->Log("Timeout during file request (4)", LL_ERROR);
						return ERR_TIMEOUT;
					}
				}

				starttime=Server->getTimeMS();

				if(protocol_version>0)
					firstpacket=true;

				hash_func.init();
				state=EReceiveState_Data;
				needs_flush=true;
			}
		}
        else
        {
			starttime=Server->getTimeMS();

			_u32 off=0;
			uchar PID=buf[0];
                        
            if( firstpacket==true)
            {
				firstpacket=false;
				if(PID==ID_COULDNT_OPEN)
				{
					if(rc>1)
					{
						memmove(dl_buf, dl_buf+1, rc-1);
						dl_off = rc-1;
					}
					return ERR_CANNOT_OPEN_FILE;
				}
				else if(PID==ID_BASE_DIR_LOST)
				{
					if(rc>1)
					{
						memmove(dl_buf, dl_buf+1, rc-1);
						dl_off = rc-1;
					}
					return ERR_BASE_DIR_LOST;
				}
				else if (PID == ID_READ_ERROR)
				{
					if (rc>1)
					{
						memmove(dl_buf, dl_buf + 1, rc - 1);
						dl_off = rc - 1;
					}
					return ERR_READ_ERROR;
				}
				else if(PID==ID_FILESIZE || PID== ID_FILESIZE_AND_EXTENTS)
				{
					size_t req_size;
					if (PID == ID_FILESIZE)
					{
						req_size = 1 + sizeof(_u64);
					}
					else
					{
						req_size = 1 + 2 * sizeof(_u64);
					}

					if(rc >= req_size)
					{
						{
							IScopedLock lock(mutex);
							is_downloading=true;
						}

						memcpy(&filesize, buf+1, sizeof(_u64) );
						filesize=little_endian(filesize);

						if (PID == ID_FILESIZE_AND_EXTENTS)
						{
							memcpy(&num_sparse_extends, buf + 1 + sizeof(_u64), sizeof(num_sparse_extends));
							num_sparse_extends = little_endian(num_sparse_extends);
						}
						else
						{
							num_sparse_extends = 0;
						}

						off = static_cast<_u32>(req_size);

						if( (filesize==0 && num_sparse_extends==0) || metadata_only)
						{
							if(rc>off)
							{
								memmove(dl_buf, dl_buf+off, rc-off);
								dl_off = rc-off;
							}
							return ERR_SUCCESS;
						}
						else if(filesize<0)
						{
							filesize = LLONG_MAX;
						}
						else if(filesize==received && num_sparse_extends==0)
						{
							if(rc>off)
							{
								memmove(dl_buf, dl_buf+off, rc-off);
								dl_off = rc-off;
							}
							return ERR_SUCCESS;
						}
						

						if(protocol_version>1)
						{
							if(filesize<next_checkpoint)
								next_checkpoint=filesize;
						}
						else
						{
							next_checkpoint=filesize;
						}

						if (num_sparse_extends > 0)
						{
							state = EReceiveState_SparseExtends;
							hash_func.init();

							Server->Log("Downloading \""+ remotefn+"\" with sparse extents (full)...", LL_DEBUG);

							resetSparseExtentsFile();

							sparse_extends_f = temporaryFileRetry();

							if (!writeFileRetry(sparse_extends_f, reinterpret_cast<char*>(&num_sparse_extends), sizeof(num_sparse_extends)))
							{
								Server->Log("Error while writing to temporary file -1", LL_ERROR);
								Reconnect();
								return ERR_ERROR;
							}
							hash_r = static_cast<_u32>(2*num_sparse_extends*sizeof(_u64) + 16);
						}
					}
					else
					{
						dl_off=rc;
						firstpacket=true;
						continue;
					}
				}
				else
				{
					if(rc>1)
					{
						memmove(dl_buf, dl_buf+1, rc-1);
						dl_off = rc-1;
					}
					Server->Log("Unknown packet id " + convert(PID), LL_ERROR);
					return ERR_ERROR;
				}
            }

			if (state == EReceiveState_SparseExtends && (_u32)rc > off)
			{
				if (hash_r > 16)
				{
					_u32 tc_hash = (std::min)((_u32)rc - off, hash_r - 16);

					hash_func.update(reinterpret_cast<unsigned char*>(&buf[off]), tc_hash);
				}

				_u32 tc = (std::min)((_u32)rc - off, hash_r);

				if (!writeFileRetry(sparse_extends_f, &buf[off], tc))
				{
					Server->Log("Error while writing to temporary file -2", LL_ERROR);
					Reconnect();
					return ERR_ERROR;
				}

				off += tc;
				hash_r -= tc;

				if (hash_r == 0)
				{
					hash_func.finalize();

					sparse_extends_f->Seek(sparse_extends_f->Size() - 16);
					std::string received_hash = sparse_extends_f->Read(16);

					if (memcmp(hash_func.raw_digest_int(), received_hash.data(), 16) != 0)
					{
						Server->Log("Error while downloading file: sparse extends hash wrong", LL_ERROR);
						Reconnect();
						return ERR_HASH;
					}

					sparse_extends_f->Seek(0);

					sparse_file.reset(new SparseFile(backing_file, sparse_extends_f, false, 1, false, -1));
					file = sparse_file.get();

					if (sparse_file->hasError())
					{
						Server->Log("Error while creating sparse file view", LL_ERROR);
						Reconnect();
						return ERR_ERROR;
					}

					if(received>0)
						file->Seek(received);

					{
						IScopedLock lock(mutex);
						sparse_bytes += sparse_file->getSparseSize();
					}

					hash_func.init();

					if (filesize == 0 || filesize == received)
					{
						if (rc>off)
						{
							memmove(dl_buf, dl_buf + off, rc - off);
							dl_off = rc - off;
						}
						return ERR_SUCCESS;
					}
					else
					{
						state = EReceiveState_Data;
					}
				}
			}

			if( state==EReceiveState_Hash && (_u32) rc > off )
			{
				_u32 tc=(std::min)((_u32)rc-off, hash_r);
				memcpy(&hash_buf[16-hash_r], &buf[off],  tc);
				off+=tc;
				hash_r-=tc;

				if(hash_r==0)
				{
					hash_func.finalize();
					if(memcmp(hash_func.raw_digest_int(), hash_buf, 16)!=0)
					{
						Server->Log("Error while downloading file: hash wrong -1", LL_ERROR);
						Reconnect();
						return ERR_HASH;
					}
					hash_func.init();
					state=EReceiveState_Data;
				}

				if(received >= filesize && state==0)
				{
					assert(received==filesize);
					if(off < rc)
					{
						memmove(dl_buf, dl_buf+off, rc-off);
						dl_off = rc-off;
					}
					return ERR_SUCCESS;
				}
			}

            if( state==EReceiveState_Data && (_u32) rc > off )
            {
				_u32 written=off;
				_u64 write_remaining=next_checkpoint-received;
				_u32 hash_off=0;
				bool c=true;
				while(c)
				{
					c=false;
					while(written<rc)
					{
						_u32 tw=(_u32)rc-written;
						if((_u64)tw>write_remaining)
							tw=(_u32)write_remaining;

						if(file==NULL)
						{
							return ERR_ERROR;
						}

						_u32 cw=file->Write(&buf[written], tw);
						hash_func.update((unsigned char*)&buf[written], cw);
						written+=cw;
						write_remaining-=cw;
						received+=cw;
						if(write_remaining==0)
							break;
						if(written<rc)
						{
							if(nofreespace_callback!=NULL
								&& !nofreespace_callback->handle_not_enough_space(file->getFilename()) )
							{
								Server->Log("Error while writing to file. No free space -2", LL_ERROR);
								Reconnect();
								return ERR_ERROR;
							}

							Server->Log("Failed to write to file... waiting...", LL_WARNING);
							Server->wait(10000);
							starttime=Server->getTimeMS();
						}
					}

					if(write_remaining==0 && protocol_version>1) 
					{
						if(next_checkpoint<filesize)
						{
							last_checkpoint=next_checkpoint;
						}
						next_checkpoint+=c_checkpoint_dist;
						if(next_checkpoint>filesize)
							next_checkpoint=filesize;

						hash_r=(_u32)rc-written;
						if(hash_r>0)
						{
							memcpy(hash_buf, &buf[written], (std::min)(hash_r, (_u32)16));

							if(received<filesize)
							{
								if(hash_r>16)
								{
									hash_r=16;
									c=true;
									write_remaining=next_checkpoint-received;
									written+=16;
								}
							}
							else if(hash_r>=16)
							{
								written+=16;
							}
						}

						hash_off+=hash_r;

						if(hash_r<16)
						{
							hash_r=16-hash_r;
							state=EReceiveState_Hash;
						}
						else
						{
							hash_func.finalize();
							if(memcmp(hash_func.raw_digest_int(), hash_buf, 16)!=0)
							{
								Server->Log("Error while downloading file: hash wrong -2", LL_ERROR);
								Reconnect();
								return ERR_HASH;
							}
							hash_func.init();
						}
					}
				}

				{
					IScopedLock lock(mutex);
					received_data_bytes+=written-off;
				}

				if( received >= filesize && state==0)
				{
					assert(received==filesize);
					if(written < rc)
					{
						memmove(dl_buf, dl_buf+written, rc-written);
						dl_off = rc-written;
					}
					return ERR_SUCCESS;
				}
            }
		}
            
	    if( Server->getTimeMS()-starttime > SERVER_TIMEOUT )
		{
			Server->Log("Server timeout in FileClient. Trying to reconnect...", LL_INFO);

			last_timeouttime = Server->getTimeMS();

			bool b = false;
			int tries = getReconnectTriesDecr();
			if (tries > 0)
			{
				b = Reconnect();
			}
			else
			{
				Server->Log("FileClient: Reconnect tries exhausted (3)", LL_INFO);
			}
			if(!b)
			{
				Server->Log("FileClient: ERR_TIMEOUT", LL_INFO);
				return ERR_TIMEOUT;
			}
			else
			{
				CWData data;
				data.addUChar( metadata_only?ID_GET_FILE_METADATA_ONLY : 
					( file_id!=0 ? ID_GET_FILE_WITH_METADATA :
					(protocol_version>1?ID_GET_FILE_RESUME_HASH:ID_GET_FILE_RESUME) ) );
				data.addString( remotefn );
				data.addString( identity );

				if(metadata_only)
				{
					data.addChar(0);
					data.addVarInt(folder_items);
					data.addVarInt(file_id);
				}
				else if(file_id!=0)
				{
					data.addChar(0);
					data.addChar(hashed?1:0);
					data.addVarInt(file_id);
					data.addChar(1);
				}

				bool resume = received > 0;

				if( protocol_version>1 )
				{
					received=last_checkpoint;
				}

				if (resume || received>0)
				{
					data.addInt64(received);
				}

				if(file!=NULL)
				{
					file->Seek(received);
				}

				while(stack.Send( tcpsock, data.getDataPtr(), data.getDataSize() )!=data.getDataSize())
				{
					Server->Log("Timeout during file request (2). Reconnecting...", LL_DEBUG);

					bool b = false;
					int tries = getReconnectTriesDecr();
					if (tries > 0)
					{
						b = Reconnect();
					}
					if (!b)
					{
						Server->Log("Timeout during file request (2)", LL_ERROR);
						return ERR_TIMEOUT;
					}
				}
				starttime=Server->getTimeMS();

				if(protocol_version>0)
					firstpacket=true;

				hash_func.init();
				state=EReceiveState_Data;
				needs_flush=true;
			}
		}

		logProgress(remotefn, filesize, received);
	}
}
        
_i64 FileClient::getTransferredBytes(void)
{
	IScopedLock lock(mutex);
	if(tcpsock!=NULL)
	{
		transferred_bytes+=tcpsock->getTransferedBytes();
		tcpsock->resetTransferedBytes();
	}
	return transferred_bytes;
}

std::string FileClient::getErrorString(_u32 ec)
{
#define DEFEC(x) case ERR_##x : return #x;
	switch(ec)
	{
	DEFEC(CONTINUE);
	DEFEC(SUCCESS);
	DEFEC(TIMEOUT);
	DEFEC(CANNOT_OPEN_FILE);
	DEFEC(SOCKET_ERROR);
	DEFEC(CONNECTED);
	DEFEC(ERROR);
	DEFEC(BASE_DIR_LOST);
	DEFEC(HASH);
	DEFEC(INT_ERROR);
	DEFEC(CONN_LOST);
	DEFEC(ERRORCODES);
	DEFEC(READ_ERROR);
	}
#undef DEFEC
	return "";
}

void FileClient::setReconnectionTimeout(unsigned int t)
{
	IScopedLock lock(mutex);
	reconnection_timeout=t;
}

_i64 FileClient::getReceivedDataBytes( bool with_sparse )
{
	IScopedLock lock(mutex);

	if (with_sparse)
	{
		return received_data_bytes + sparse_bytes;
	}
	else
	{
		return received_data_bytes;
	}
}

void FileClient::resetReceivedDataBytes(bool with_sparse)
{
	IScopedLock lock(mutex);
	received_data_bytes=0;

	if (with_sparse)
	{
		sparse_bytes = 0;
	}
}

void FileClient::setQueueCallback( QueueCallback* cb )
{
	queue_callback = cb;
}

_u32 FileClient::fillQueue()
{
	if(queue_callback==NULL)
	{
		if(needs_flush)
		{
			needs_flush=false;
			return Flush();
		}

		return ERR_SUCCESS;
	}

	if(queued.size()>queuedFilesLow)
	{
		if (needs_flush)
		{
			needs_flush = false;
			return Flush();
		}

		return ERR_SUCCESS;
	}

	bool needs_send_flush=false;

	std::vector<SQueueItem> queued_files;
	int64 queue_starttime = Server->getTimeMS();

	while(queued.size()<maxQueuedFiles
		&& Server->getTimeMS()-queue_starttime<10000)
	{
		if(!tcpsock->isWritable())
		{
			if (needs_flush)
			{
				needs_flush = false;
				return Flush();
			}

			return ERR_SUCCESS;
		}

		MetadataQueue metadata_queue = MetadataQueue_Data;
		size_t folder_items = 0;
		bool finish_script=false;
		int64 file_id;
		std::string queue_fn = queue_callback->getQueuedFileFull(metadata_queue, folder_items, finish_script, file_id);

		if(queue_fn.empty())
		{
			if(needs_flush)
			{
				needs_flush=false;
				needs_send_flush=false;
				return Flush();
			}

			return ERR_SUCCESS;
		}

		CWData data;
		if(metadata_queue==MetadataQueue_Data)
		{
			if(file_id==0)
			{
				data.addUChar( protocol_version>1?ID_GET_FILE_RESUME_HASH:ID_GET_FILE_RESUME );
			}
			else
			{
				data.addUChar( ID_GET_FILE_WITH_METADATA );
			}
		}
		else if(metadata_queue == MetadataQueue_MetadataAndHash)
		{
			data.addUChar( ID_FILE_HASH_AND_METADATA );
		}
		else if(metadata_queue == MetadataQueue_Metadata)
		{
			data.addUChar(ID_GET_FILE_METADATA_ONLY);
		}
		data.addString( queue_fn );
		data.addString( identity );

		if(metadata_queue == MetadataQueue_Metadata)
		{
			data.addChar(0);
			data.addVarInt(folder_items);
			data.addVarInt(file_id);
		}
		else if(file_id!=0)
		{
			data.addChar(0);
			data.addChar(protocol_version>1);
			data.addVarInt(file_id);
			data.addChar(1);
		}

		needs_send_flush=true;
		if(stack.Send( tcpsock, data.getDataPtr(), data.getDataSize(), c_default_timeout, false)!=data.getDataSize())
		{
			Server->Log("Queueing file failed", LL_DEBUG);

			for (size_t i = 0; i<queued_files.size(); ++i)
			{
				queue_callback->unqueueFileFull(queued_files[i].fn, queued_files[i].finish_script);
			}

			queue_callback->unqueueFileFull(queue_fn, finish_script);

			return ERR_TIMEOUT;
		}

		queued.push_back(SQueueItem(queue_fn, finish_script));
		queued_files.push_back(SQueueItem(queue_fn, finish_script));
		needs_flush=true;
	}

	if (needs_flush)
	{
		needs_flush = false;
		return Flush();
	}
	else if(needs_send_flush)
	{
		if(!tcpsock->Flush(c_default_timeout))
		{
			Server->Log("Flushing failed after queueing files", LL_DEBUG);
			for(size_t i=0;i<queued_files.size();++i)
			{
				queue_callback->unqueueFileFull(queued_files[i].fn, queued_files[i].finish_script);
			}
			return ERR_TIMEOUT;
		}
	}

	return ERR_SUCCESS;
}

void FileClient::logProgress(const std::string& remotefn, _u64 filesize, _u64 received)
{
	int64 ct = Server->getTimeMS();
	if(filesize>0 && (last_progress_log==0 ||
		ct-last_progress_log>60000) )
	{
		int64 new_transferred=getTransferredBytes();
		if( last_transferred_bytes!=0 &&
			last_progress_log!=0 )
		{
			int64 tranferred = new_transferred - last_transferred_bytes;
			int64 speed_bps = tranferred*1000 / (ct-last_progress_log);

			if(tranferred>0 && progress_log_callback)
			{
				progress_log_callback->log_progress(remotefn,
					filesize, received, speed_bps);
			}
		}

		last_transferred_bytes = new_transferred;
		last_progress_log = ct;
	}	
}

bool FileClient::alreadyHasAddrv4(sockaddr_in addr)
{
	for (size_t i = 0; i < udpsocks.size(); ++i)
	{
		if (!udpsocks[i].is_ipv6
			&& addr.sin_addr.s_addr == udpsocks[i].addr_ipv4.sin_addr.s_addr)
			return true;
	}
	return false;
}

bool FileClient::alreadyHasAddrv6(sockaddr_in6 addr)
{
	for (size_t i = 0; i < udpsocks.size(); ++i)
	{
		if (udpsocks[i].is_ipv6
			&& memcmp(&addr.sin6_addr, &udpsocks[i].addr_ipv6.sin6_addr, sizeof(addr.sin6_addr))==0)
			return true;
	}
	return false;
}

std::string FileClient::ipToString(SSocket& s)
{
	if (!s.is_ipv6)
	{
		char str[INET_ADDRSTRLEN];
		inet_ntop(AF_INET, &(s.addr_ipv4.sin_addr), str, INET_ADDRSTRLEN);
		return str;
	}
	else
	{
		char str[INET6_ADDRSTRLEN];
		inet_ntop(AF_INET6, &(s.addr_ipv6.sin6_addr), str, INET6_ADDRSTRLEN);
		return str;
	}
}

IFile * FileClient::temporaryFileRetry()
{
	while (true)
	{
		IFile* ret = Server->openTemporaryFile();

		if (ret == NULL)
		{
			Server->Log("Error opening temporary file in FileClient. Retrying...", LL_WARNING);
			Server->wait(10000);
		}
		else
		{
			return ret;
		}
	}
}

bool FileClient::writeFileRetry(IFile * f, const char * buf, _u32 bsize)
{
	_u32 w = 0;
	while (w < bsize)
	{
		bool has_error = false;
		w += f->Write(buf + w, bsize - w, &has_error);

		if (has_error)
		{
			return false;
		}

		if (w < bsize)
		{
			Server->Log("Error writing to file " + f->getFilename() + ". Retrying...", LL_WARNING);
			Server->wait(10000);
		}
	}
	return true;
}

void FileClient::setReconnectTries(int tries)
{
	IScopedLock lock(mutex);
	reconnect_tries = tries;
}

void FileClient::setAddChecksum(bool add_request_checksums)
{
	stack.setAddChecksum(add_request_checksums);
}

void FileClient::setProgressLogCallback( ProgressLogCallback* cb )
{
	progress_log_callback = cb;
}

FileClient::ProgressLogCallback * FileClient::getProgressLogCallback()
{
	return progress_log_callback;
}

void FileClient::setNoFreeSpaceCallback(FileClient::NoFreeSpaceCallback * cb)
{
	nofreespace_callback = cb;
}

/*_u32 FileClient::GetFileHashAndMetadata( std::string remotefn, std::string& hash, std::string& permissions, int64& filesize, int64& created, int64& modified )
{
	if (tcpsock == NULL)
		return ERR_ERROR;

	if(queued.empty())
	{
		CWData data;
		data.addUChar( ID_GET_FILE_HASH_AND_METADATA );
		data.addString( remotefn );
		data.addString( identity );

		if(stack.Send( tcpsock, data.getDataPtr(), data.getDataSize() )!=data.getDataSize())
		{
			Server->Log("Timeout during file hash request (1)", LL_ERROR);
			return ERR_TIMEOUT;
		}
	}
	else
	{
		assert(queued.front().fn == remotefn);
		assert(!queued.front().finish_script);
		queued.pop_front();
	}


	bool firstpacket=true;
	setReconnectTries(50);
	unsigned short metadata_size;
	std::string metadata;
	size_t metadata_pos=0;

	while(true)
	{        
		fillQueue();

		size_t rc;
		if(tcpsock->isReadable() || dl_off==0 || 
			(firstpacket && dl_buf[0]==ID_GET_FILE_HASH_AND_METADATA && dl_off<1+sizeof(unsigned short) ) )
		{
			rc = tcpsock->Read(&dl_buf[dl_off], BUFFERSIZE-dl_off, 120000)+dl_off;
		}
		else
		{
			rc = dl_off;
		}
		dl_off=0;

		if( rc==0 )
		{
			Server->Log("Server timeout (2) in FileClient while getting hash and metadata", LL_DEBUG);
			int tries = getReconnectTriesDecr();
			bool b = false;
			if (tries > 0)
			{
				b = Reconnect();
			}
			if(!b )
			{
				Server->Log("FileClient: ERR_TIMEOUT", LL_INFO);
				return ERR_TIMEOUT;
			}
			else
			{
				CWData data;
				data.addUChar( ID_GET_FILE_HASH_AND_METADATA );
				data.addString( remotefn );
				data.addString( identity );

				rc=stack.Send( tcpsock, data.getDataPtr(), data.getDataSize() );
				if(rc==0)
				{
					Server->Log("FileClient: Error sending request for hash and metadata", LL_INFO);
				}
				starttime=Server->getTimeMS();

				firstpacket=true;
				needs_flush=true;
			}
		}
		else
		{
			starttime=Server->getTimeMS();

			_u32 off=0;
			uchar PID=dl_buf[0];

			if( firstpacket==true)
			{
				firstpacket=false;
				if(PID==ID_COULDNT_OPEN)
				{
					if(rc>1)
					{
						memmove(dl_buf, dl_buf+1, rc-1);
						dl_off = rc-1;
					}
					return ERR_CANNOT_OPEN_FILE;
				}
				else if(PID==ID_BASE_DIR_LOST)
				{
					if(rc>1)
					{
						memmove(dl_buf, dl_buf+1, rc-1);
						dl_off = rc-1;
					}
					return ERR_BASE_DIR_LOST;
				}
				else if(PID==ID_FILE_HASH_AND_METADATA)
				{
					if(rc >= 1+sizeof(unsigned short))
					{
						memcpy(&metadata_size, dl_buf+1, sizeof(metadata_size) );
						off=1+sizeof(metadata_size);

						metadata.resize(metadata_size);

						if( metadata_size==0 )
						{
							if(rc>off)
							{
								memmove(dl_buf, dl_buf+off, rc-off);
								dl_off = rc-off;
							}
							return ERR_ERROR;
						}
					}
					else
					{
						dl_off=rc;
						firstpacket=true;
						continue;
					}
				}
				else
				{
					if(rc>1)
					{
						memmove(dl_buf, dl_buf+1, rc-1);
						dl_off = rc-1;
					}
					return ERR_ERROR;
				}

				if(rc>off)
				{
					size_t toread=rc-off;
					if(toread>metadata_size)
					{
						toread=metadata_size;
					}

					memcpy(&metadata[metadata_pos], &dl_buf[off], toread);
					metadata_pos+=toread;
					
					off+=static_cast<_u32>(toread);

					if(rc>off)
					{
						memmove(dl_buf, dl_buf+off, rc-off);
						dl_off = rc-off;
					}

					if(metadata_pos==metadata.size())
					{
						CRData data(&metadata);

						if(!data.getStr(&metadata))
						{
							return ERR_ERROR;
						}

						if(!data.getStr(&permissions))
						{
							return ERR_ERROR;
						}

						if(!data.getInt64(&filesize))
						{
							return ERR_ERROR;
						}

						if(!data.getInt64(&modified))
						{
							return ERR_ERROR;
						}

						if(!data.getInt64(&created))
						{
							return ERR_ERROR;
						}

						return ERR_SUCCESS;
					}
				}
			}
		}

		if( Server->getTimeMS()-starttime > SERVER_TIMEOUT )
		{
			Server->Log("Server timeout in FileClient while downloading hash and metadata. Trying to reconnect...", LL_INFO);
			int tries = getReconnectTriesDecr();
			bool b = false;
			if (tries > 0)
			{
				b = Reconnect();
			}			
			if(!b)
			{
				Server->Log("FileClient: ERR_TIMEOUT", LL_INFO);
				return ERR_TIMEOUT;
			}
			else
			{
				CWData data;
				data.addUChar( ID_GET_FILE_HASH_AND_METADATA );
				data.addString( remotefn );
				data.addString( identity );

				rc=stack.Send( tcpsock, data.getDataPtr(), data.getDataSize() );
				if(rc==0)
				{
					Server->Log("FileClient: Error sending request for hash and metadata", LL_INFO);
				}
				starttime=Server->getTimeMS();

				firstpacket=true;
				needs_flush=true;
			}
		}
	}
}*/

_u32 FileClient::InformMetadataStreamEnd( const std::string& server_token, int tries)
{
	if (tcpsock == NULL)
		return ERR_ERROR;

	setReconnectTries(tries);

	CWData data;
	data.addUChar( ID_INFORM_METADATA_STREAM_END );
	data.addString( identity );
	data.addString(server_token);

	if(stack.Send( tcpsock, data.getDataPtr(), data.getDataSize() )!=data.getDataSize())
	{
		Server->Log("Timeout during sending metadata stream end (1)", LL_ERROR);
		return ERR_TIMEOUT;
	}

	while(true)
	{
		size_t rc = tcpsock->Read(dl_buf, 1, 120000);

		if(rc==0)
		{
			Server->Log("Server timeout (2) in FileClient sending metadata stream end", LL_DEBUG);
			int tries = getReconnectTriesDecr();
			bool b = false;
			if (tries > 0)
			{
				b = Reconnect();
			}
			if(!b)
			{
				Server->Log("FileClient: ERR_TIMEOUT (metadata stream)", LL_INFO);
				return ERR_TIMEOUT;
			}
			else
			{
				CWData data;
				data.addUChar( ID_INFORM_METADATA_STREAM_END );
				data.addString( identity );
				data.addString(server_token);

				rc=stack.Send( tcpsock, data.getDataPtr(), data.getDataSize() );
				if(rc==0)
				{
					Server->Log("FileClient: Error sending metadata stream end", LL_INFO);
				}
				starttime=Server->getTimeMS();
			}
		}
		else
		{
			if(*dl_buf==ID_PONG)
			{
				return ERR_SUCCESS;
			}
			else
			{
				return ERR_ERROR;
			}
		}
	}
}

_u32 FileClient::StopPhashLoad(const std::string & server_token, const std::string & phash_fn, int tries)
{
	if (tcpsock == NULL)
		return ERR_ERROR;

	setReconnectTries(tries);

	CWData data;
	data.addUChar(ID_STOP_PHASH);
	data.addString(identity);
	data.addString(server_token);
	data.addString(phash_fn);

	if (stack.Send(tcpsock, data.getDataPtr(), data.getDataSize()) != data.getDataSize())
	{
		Server->Log("Timeout during sending phash stream end (1)", LL_ERROR);
		return ERR_TIMEOUT;
	}

	while (true)
	{
		size_t rc = tcpsock->Read(dl_buf, 1, 120000);

		if (rc == 0)
		{
			Server->Log("Server timeout (2) in FileClient sending phash stream end", LL_DEBUG);
			int tries = getReconnectTriesDecr();
			bool b = false;
			if (tries > 0)
			{
				b = Reconnect();
			}
			if (!b)
			{
				Server->Log("FileClient: ERR_TIMEOUT (phash stream)", LL_INFO);
				return ERR_TIMEOUT;
			}
			else
			{
				CWData data;
				data.addUChar(ID_INFORM_METADATA_STREAM_END);
				data.addString(identity);
				data.addString(server_token);

				rc = stack.Send(tcpsock, data.getDataPtr(), data.getDataSize());
				if (rc == 0)
				{
					Server->Log("FileClient: Error sending phash stream end", LL_INFO);
				}
				starttime = Server->getTimeMS();
			}
		}
		else
		{
			if (*dl_buf == ID_PONG)
			{
				return ERR_SUCCESS;
			}
			else
			{
				return ERR_ERROR;
			}
		}
	}
}

_u32 FileClient::FinishScript(std::string remotefn)
{
	if (tcpsock == NULL)
		return ERR_ERROR;


	if(queued.empty())
	{
		CWData data;
		data.addUChar( ID_SCRIPT_FINISH );
		data.addString( identity );
		data.addString(remotefn);

		int tries=5000;

		if(stack.Send( tcpsock, data.getDataPtr(), data.getDataSize() )!=data.getDataSize())
		{
			Server->Log("Timeout during sending finish script (1)", LL_ERROR);
			return ERR_TIMEOUT;
		}
	}
	else
	{
		assert(queued.front().fn == remotefn);
		assert(queued.front().finish_script);
		queued.pop_front();
	}

	int tries=20;

	while(true)
	{
		size_t rc = tcpsock->Read(dl_buf, 1, 120000);

		if(rc==0)
		{
			Server->Log("Server timeout (2) in FileClient sending finish script", LL_DEBUG);
			bool b=Reconnect();
			--tries;
			if(!b || tries<=0 )
			{
				Server->Log("FileClient: ERR_TIMEOUT (finish script)", LL_INFO);
				return ERR_TIMEOUT;
			}
			else
			{
				CWData data;
				data.addUChar( ID_SCRIPT_FINISH );
				data.addString( identity );
				data.addString(remotefn);

				rc=stack.Send( tcpsock, data.getDataPtr(), data.getDataSize() );
				if(rc==0)
				{
					Server->Log("FileClient: Error sending metadata stream end", LL_INFO);
				}
				starttime=Server->getTimeMS();
			}
		}
		else
		{
			if(*dl_buf==ID_PONG)
			{
				return ERR_SUCCESS;
			}
			else
			{
				return ERR_ERROR;
			}
		}
	}
}

_u32 FileClient::Flush()
{
	if(tcpsock==NULL)
		return ERR_ERROR;

	Server->Log("Flushing FileClient...", LL_DEBUG);

	CWData data;
	data.addUChar(ID_FLUSH_SOCKET);

	if(stack.Send( tcpsock, data.getDataPtr(), data.getDataSize() )!=data.getDataSize())
	{
		Server->Log("Timeout during flush request", LL_ERROR);
		return ERR_TIMEOUT;
	}

	return ERR_SUCCESS;
}

_i64 FileClient::getRealTransferredBytes()
{
	if(tcpsock!=NULL)
	{
		return real_transferred_bytes+=tcpsock->getRealTransferredBytes();
	}
	return real_transferred_bytes;
}

void FileClient::Shutdown()
{
	IScopedLock lock(mutex);
	if(tcpsock!=NULL)
	{
		tcpsock->shutdown();
	}
	reconnect_tries = 0;
	reconnection_timeout = 10000;
}

bool FileClient::isDownloading()
{
	IScopedLock lock(mutex);
	return is_downloading;
}

IFile * FileClient::releaseSparseExtendsFile()
{
	IFile* ret = sparse_extends_f;
	sparse_extends_f = NULL;
	return ret;
}

void FileClient::resetSparseExtentsFile()
{
	if (sparse_extends_f != NULL)
	{
		std::string tmpfn = sparse_extends_f->getFilename();
		Server->destroy(sparse_extends_f);
		Server->deleteFile(tmpfn);
		sparse_extends_f = NULL;
	}
}

std::string FileClient::SAddrHint::toString() const
{
	if (!is_ipv6)
	{
		char str[INET_ADDRSTRLEN];
		inet_ntop(AF_INET, (void*)&addr_ipv4, str, INET_ADDRSTRLEN);
		return str;
	}
	else
	{
		char str[INET6_ADDRSTRLEN];
		sockaddr_in6 saddr = {};
		saddr.sin6_family = AF_INET6;
		saddr.sin6_port = htons(UDP_SOURCE_PORT);
		memcpy(&saddr.sin6_addr, addr_ipv6, sizeof(addr_ipv6));
		saddr.sin6_scope_id = zone;
		if (getnameinfo(reinterpret_cast<sockaddr*>(&saddr), sizeof(saddr), 
			str, sizeof(str), NULL, 0, NI_NUMERICHOST) != 0)
		{
			return std::string();
		}
		return str;
	}
}
