<commit_before>#include "stdafx.h"
#include "Wind_API.h"

#include "util.h"
#include "WindEvent.h"

#include "kdb+.util/K_ptr.h"
#include "kdb+.util/type_convert.h"
#include "win32.util/SocketPair.h"
#include <iostream>
#include <mutex>	//C++11
#include <map>

namespace Wind {
	namespace pubsub {

		bool prepare() {
			return SockPair::prepare();
		}

		bool finalize() {
			return SockPair::finalize();
		}

		// This is essentially a concurrent map...
		//NOTE: All std::map<> interface are _not_ protected! Use only the member functions wrapped below.
		class CallbackRegistry : private std::map<::WQID, std::string> {
		public:
			std::string& operator[](::WQID qid) {
				lock_guard guard(lock_);
				return parent::operator[](qid);
			}
			size_type erase(::WQID qid) {
				lock_guard guard(lock_);
				return parent::erase(qid);
			}
		private:
			typedef std::map<::WQID, std::string> parent;
			typedef std::lock_guard<std::mutex> lock_guard;
			mutable std::mutex lock_;
		};

		static CallbackRegistry REGISTRY;
		std::size_t const CLIENT = 0;
		std::size_t const SERVER = 1;

		// Data processor (executed within q's main thread)
		K invokeCallback(I socket) {
			static_assert(sizeof(::SOCKET) == sizeof(I), "SOCKET vs I: type mismatch!");
			::SOCKET sock = socket;
			assert(sock != INVALID_SOCKET);

			// Receive (WQID, len, serialized_K) from subscription thread
#			define RECV_CHECK(expectedSize, errorMsg)	\
			if (recvd != (expectedSize)) {	\
				std::cerr << "<recv> " << (errorMsg) << ": " << recvd << " < " << (expectedSize) << std::endl;	\
				return K_NIL;	\
			}
			::WQID qid = 0;
			int recvd = ::recv(sock, reinterpret_cast<char*>(&qid), sizeof(::WQID), 0);
			RECV_CHECK(sizeof(::WQID), "WQID incomplete");
			std::size_t len = 0;
			recvd = ::recv(sock, reinterpret_cast<char*>(&len), sizeof(len), 0);
			RECV_CHECK(sizeof(len), "size incomplete");
			if (len > static_cast<std::size_t>(std::numeric_limits<int>::max())) {
				std::cerr << "<recv> serialized data (" << len << ") > 2G" << std::endl;
				return K_NIL;
			}
			q::K_ptr serialized(ktn(KB, len));
			std::memset(kG(serialized.get()), 0, len);
			recvd = ::recv(sock, reinterpret_cast<char*>(kG(serialized.get())), len, 0);
			RECV_CHECK(len, "data incomplete");
#			undef RECV_CHECK

			// Deserialize K object
#			if KX_USE_OKX
			//@ref https://groups.google.com/forum/#!topic/personal-kdbplus/pjsugT7590A
			if (!okx(serialized.get())) {
				std::cerr << "<recv> bad data: ["
					<< util::hexBytes(kG(serialized.get()), static_cast<std::size_t>(serialized->n))
					<< ']' << std::endl;
				return K_NIL;
			}
#			endif
			q::K_ptr result(d9(serialized.get()));

			// Identify the origial query and callback
			std::string const callback = REGISTRY[qid];
			if (callback.empty()) {
				std::cerr << "unknown WQID: " << qid << std::endl;
				return K_NIL;
			}

			static_assert(std::is_same<::WQID, J>::value, "WQID data type mismatch");
			q::K_ptr output(k(0, const_cast<S>(callback.c_str()), kj(qid), result.release(), K_NIL));
			if (output->t == -128) {
				std::cerr << "<q> '" << output->s << std::endl;
			}
			return output.release();
		}<commit_msg>Bugfix: large WSQ subscriptions' callback message might get fragmented.<commit_after>#include "stdafx.h"
#include "Wind_API.h"

#include "util.h"
#include "WindEvent.h"

#include "kdb+.util/K_ptr.h"
#include "kdb+.util/type_convert.h"
#include "win32.util/SocketPair.h"
#include <iostream>
#include <mutex>	//C++11
#include <map>

namespace Wind {
	namespace pubsub {

		bool prepare() {
			return SockPair::prepare();
		}

		bool finalize() {
			return SockPair::finalize();
		}

		// This is essentially a concurrent map...
		//NOTE: All std::map<> interface are _not_ protected! Use only the member functions wrapped below.
		class CallbackRegistry : private std::map<::WQID, std::string> {
		public:
			std::string& operator[](::WQID qid) {
				lock_guard guard(lock_);
				return parent::operator[](qid);
			}
			size_type erase(::WQID qid) {
				lock_guard guard(lock_);
				return parent::erase(qid);
			}
		private:
			typedef std::map<::WQID, std::string> parent;
			typedef std::lock_guard<std::mutex> lock_guard;
			mutable std::mutex lock_;
		};

		static CallbackRegistry REGISTRY;
		std::size_t const CLIENT = 0;
		std::size_t const SERVER = 1;

		// Data processor (executed within q's main thread)
		K invokeCallback(I socket) {
			static_assert(sizeof(::SOCKET) == sizeof(I), "SOCKET vs I: type mismatch!");
			::SOCKET sock = socket;
			assert(sock != INVALID_SOCKET);

			// Receive (WQID, len, serialized_K) from subscription thread
#			define RECV_CHECK(expectedSize, errorMsg)	\
			if (recvd != (expectedSize)) {	\
				std::cerr << "<recv> " << (errorMsg) << ": " << recvd << " < " << (expectedSize) << std::endl;	\
				return K_NIL;	\
			}
			::WQID qid = 0;
			int recvd = ::recv(sock, reinterpret_cast<char*>(&qid), sizeof(::WQID), MSG_WAITALL);
			RECV_CHECK(sizeof(::WQID), "WQID incomplete");
			std::size_t len = 0;
			recvd = ::recv(sock, reinterpret_cast<char*>(&len), sizeof(len), MSG_WAITALL);
			RECV_CHECK(sizeof(len), "size incomplete");
			if (len > static_cast<std::size_t>(std::numeric_limits<int>::max())) {
				std::cerr << "<recv> serialized data (" << len << ") > 2G" << std::endl;
				return K_NIL;
			}
			q::K_ptr serialized(ktn(KB, len));
			std::memset(kG(serialized.get()), 0, len);
			recvd = ::recv(sock, reinterpret_cast<char*>(kG(serialized.get())), len, MSG_WAITALL);
			RECV_CHECK(len, "data incomplete");
#			undef RECV_CHECK

			// Deserialize K object
#			if KX_USE_OKX
			//@ref https://groups.google.com/forum/#!topic/personal-kdbplus/pjsugT7590A
			if (!okx(serialized.get())) {
				std::cerr << "<recv> bad data: ["
					<< util::hexBytes(kG(serialized.get()), static_cast<std::size_t>(serialized->n))
					<< ']' << std::endl;
				return K_NIL;
			}
#			endif
			q::K_ptr result(d9(serialized.get()));

			// Identify the origial query and callback
			std::string const callback = REGISTRY[qid];
			if (callback.empty()) {
				std::cerr << "unknown WQID: " << qid << std::endl;
				return K_NIL;
			}

			static_assert(std::is_same<::WQID, J>::value, "WQID data type mismatch");
			q::K_ptr output(k(0, const_cast<S>(callback.c_str()), kj(qid), result.release(), K_NIL));
			if (output->t == -128) {
				std::cerr << "<q> '" << output->s << std::endl;
			}
			return output.release();
		}
{'fraction_non_alphanumeric': 0.18256937948238228, 'fraction_numerical': 0.006392266916120985, 'mean_word_length': 4.035321821036106, 'pattern_counts': {'":': 2, '<': 127, '<?xml version=': 0, '>': 51, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1975466', 'n_tokens_mistral': 2732, 'n_tokens_neox': 2344, 'n_words': 508}