Новая библиотека для Redis

.clang-tidy

@@ -223,7 +223,7 @@ CheckOptions:
   - key: readability-identifier-naming.ProtectedMemberCase
     value: camelBack
   - key: readability-identifier-naming.ProtectedMemberPrefix
-    value: ''
+    value: 'm_'
   - key: readability-identifier-naming.ProtectedMemberSuffix
     value: ''
   - key: readability-identifier-naming.ProtectedMethodCase

CMakeLists.txt

@@ -50,7 +50,7 @@ add_subdirectory(src/myx/redis)
 if(MYXLIB_BUILD_EXAMPLES OR MYXLIB_BUILD_EXAMPLES_HO)
   add_subdirectory(examples/filesystem)
   add_subdirectory(examples/qt)
-  add_subdirectory(examples/redis)
+#  add_subdirectory(examples/redis)
 endif()
 
 # Документация

cmake/cmlib

@@ -1 +1 @@
-Subproject commit 1eaded9031726f539cd5fd4a44d874d7bd8d458d
+Subproject commit 2ea38c32c8d6aa7ccdbdf84cff1d0b0d50cc2fb3

src/myx/redis.old/CMakeLists.txt

@@ -0,0 +1,72 @@
+# Название основной цели и имя библиотеки в текущем каталоге
+set(TRGT redis)
+
+# cmake-format: off
+# Список файлов исходных текстов
+set(TRGT_cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/client.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/lexer.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/parser.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/request.cpp)
+
+# Список заголовочных файлов (используется для установки)
+set(TRGT_moc_hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/client.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/lexer.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/parser.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/request.hpp)
+
+set(TRGT_hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/reply.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/client-inl.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/lexer-inl.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/parser-inl.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/request-inl.hpp)
+
+set(TRGT_headers ${TRGT_moc_hpp} ${TRGT_hpp})
+# cmake-format: on
+
+qt5_wrap_cpp(TRGT_moc_cpp ${TRGT_moc_hpp})
+
+add_library(${TRGT}-header-only INTERFACE)
+target_include_directories(
+  ${TRGT}-header-only SYSTEM INTERFACE "$<BUILD_INTERFACE:${CMAKE_CURRENT_LIST_DIR}/include>"
+                                       "$<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}>")
+
+if(MYXLIB_BUILD_LIBRARIES)
+  add_common_library(${TRGT} OUTPUT_NAME myx-${TRGT} SOURCES ${TRGT_moc_cpp} ${TRGT_cpp} ${TRGT_headers})
+  common_target_properties(${TRGT})
+
+  # Создание цели для проверки утилитой clang-tidy
+  add_clang_tidy_check(${TRGT} ${TRGT_cpp} ${TRGT_headers})
+
+  # Создание цели для проверки утилитой clang-analyze
+  add_clang_analyze_check(${TRGT} ${TRGT_cpp} ${TRGT_headers})
+
+  # Создание цели для проверки утилитой clazy
+  add_clazy_check(${TRGT} ${TRGT_cpp} ${TRGT_headers})
+
+  # Создание цели для проверки утилитой pvs-studio
+  add_pvs_check(${TRGT})
+
+  # Создание цели для автоматического форматирования кода
+  add_format_sources(${TRGT} ${TRGT_cpp} ${TRGT_headers})
+
+  target_compile_definitions(${TRGT} PUBLIC MYXLIB_BUILD_LIBRARIES)
+  target_include_directories(${TRGT} SYSTEM PUBLIC ${Qt5Core_INCLUDE_DIRS})
+  target_include_directories(${TRGT} SYSTEM PUBLIC ${Qt5Network_INCLUDE_DIRS})
+  target_include_directories(${TRGT} SYSTEM PRIVATE ${CMAKE_SOURCE_DIR}/src)
+
+  cotire(${TRGT})
+  install(TARGETS ${TRGT}_static COMPONENT libs-dev ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR})
+  if(BUILD_SHARED_LIBS)
+    install(TARGETS ${TRGT}_shared COMPONENT main LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR})
+  endif()
+endif()
+
+generate_pkgconfig(myx-${TRGT} COMPONENT base-dev INSTALL_LIBRARY ${MYXLIB_BUILD_LIBRARIES})
+install(FILES ${TRGT_headers} COMPONENT base-dev DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}/${PROJECT_NAME}/${TRGT})
+
+# Цель, используемая только для установки заголовочных файлов без компиляции проекта
+add_custom_target(${TRGT}-install-headers COMMAND ${CMAKE_COMMAND} -DCMAKE_INSTALL_COMPONENT=base-dev -P
+                                                  "${CMAKE_BINARY_DIR}/cmake_install.cmake")

src/myx/redis/CMakeLists.txt

@@ -4,24 +4,20 @@ set(TRGT redis)
 # cmake-format: off
 # Список файлов исходных текстов
 set(TRGT_cpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/client.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/base.cpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/lexer.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/pubsub.cpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/parser.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/containers.cpp)
-    ${CMAKE_CURRENT_SOURCE_DIR}/request.cpp)
 
-# Список заголовочных файлов (используется для установки)
+# Список заголовочных файлов
 set(TRGT_moc_hpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/client.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/base.hpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/lexer.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/pubsub.hpp)
-    ${CMAKE_CURRENT_SOURCE_DIR}/parser.hpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/request.hpp)
 
 set(TRGT_hpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/reply.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/base-inl.hpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/client-inl.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/containers.hpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/lexer-inl.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/containers-inl.hpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/parser-inl.hpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/pubsub-inl.hpp)
-    ${CMAKE_CURRENT_SOURCE_DIR}/request-inl.hpp)
 
 set(TRGT_headers ${TRGT_moc_hpp} ${TRGT_hpp})
 # cmake-format: on

src/myx/redis/base-inl.hpp

@@ -0,0 +1,173 @@
+#ifndef MYX_REDIS_BASE_INL_HPP_
+#define MYX_REDIS_BASE_INL_HPP_
+
+#pragma once
+
+#ifndef MYXLIB_HEADER_ONLY
+#include <myx/redis/base.hpp>
+#endif
+
+#include <myx/base/config.hpp>
+
+#include <QTimer>
+
+namespace myx {
+
+namespace redis {
+
+static const QByteArray Separator { "\r\n" }; //!< Строковый разделитель слов во фразе
+
+/*!
+ * \brief Создает объект класса
+ * \param host Имя хоста сервера
+ * \param port Номер порта сервера
+ * \param parent Родительский объект
+ */
+inline Base::Base( QObject* parent ) :
+	QObject     { parent },
+	m_readTimer ( new QTimer( this ) ),
+	m_writeTimer( new QTimer( this ) )
+{
+	m_readTimer->setSingleShot( true );
+	m_writeTimer->setSingleShot( true );
+	m_readTimer->setInterval( rwTimeout() );
+	m_writeTimer->setInterval( rwTimeout() );
+	connect( m_readTimer, &QTimer::timeout, this, [this]() {
+		if ( ( ( m_connectionFlags & kConnected ) != 0 ) && ( ( m_connectionFlags & kRead ) != 0 ) )
+		{
+			m_connectionFlags -= kRead;
+			Q_EMIT connectionStateChanged( m_connectionFlags );
+		}
+	} );
+	connect( m_writeTimer, &QTimer::timeout, this, [this]() {
+		if ( ( ( m_connectionFlags & kConnected ) != 0 ) && ( ( m_connectionFlags & kWrite ) != 0 ) )
+		{
+			m_connectionFlags -= kWrite;
+			Q_EMIT connectionStateChanged( m_connectionFlags );
+		}
+	} );
+}
+
+
+inline QString Base::host() const
+{
+	return( m_host );
+}
+
+
+/*!
+ * \brief Задает адрес хоста
+ * \param host Адрес
+ */
+inline void Base::setHost( const QString& host )
+{
+	m_host = host;
+}
+
+
+inline quint16 Base::port() const
+{
+	return( m_port );
+}
+
+
+inline void Base::setPort( quint16 port )
+{
+	m_port = port;
+}
+
+
+inline int Base::connectionFlags() const
+{
+	return( m_connectionFlags );
+}
+
+
+inline void Base::setConnectionFlags( int connectionFlags )
+{
+	m_connectionFlags = connectionFlags;
+}
+
+
+/*!
+ * \brief Возвращает признак установленного соединения
+ * \return Признак установленного соединения
+ */
+inline bool Base::isConnected() const
+{
+	return( ( m_connectionFlags & kConnected ) != 0 );
+}
+
+
+/// *!
+// * \brief Извлекает длину токена из буфера данных
+// * \param pos Позиция в буфере
+// * \param buffer Буфер данных
+// */
+// inline int Base::fetchLength( QByteArray& buffer, int& pos )
+// {
+//	if ( ( buffer[pos] != '$' ) && ( buffer[pos] != '*' ) && ( buffer[pos] != ':' ) )
+//	{
+//		buffer.clear();
+//		pos = 0;
+//		return( -1 );
+//	}
+//	pos++;
+//	QByteArray text;
+//	while ( pos < buffer.length() )
+//	{
+//		if ( QChar( buffer[pos] ).isNumber() )
+//		{
+//			text += buffer[pos++];
+//		}
+//		else
+//		{
+//			pos += Separator.length();
+//			return( text.toInt() );
+//		}
+//	}
+//	return( -1 );
+// } // Base::fetchLength
+
+
+/*!
+ * \brief Добавляет к состоянию флаг Read
+ */
+inline void Base::onReadyRead()
+{
+	if ( ( ( m_connectionFlags & kConnected ) != 0 ) && ( ( m_connectionFlags & kRead ) == 0 ) )
+	{
+		m_connectionFlags += kRead;
+		Q_EMIT connectionStateChanged( m_connectionFlags );
+	}
+	m_readTimer->start();
+}
+
+
+/*!
+ * \brief Добавляет к состоянию флаг Read
+ */
+inline void Base::onBytesWritten()
+{
+	if ( ( ( m_connectionFlags & kConnected ) != 0 ) && ( ( m_connectionFlags & kWrite ) == 0 ) )
+	{
+		m_connectionFlags += kWrite;
+		Q_EMIT connectionStateChanged( m_connectionFlags );
+	}
+	m_writeTimer->start();
+}
+
+
+/*!
+ * \brief Выставляет флаг ошибки соединения
+ */
+inline void Base::onSocketError()
+{
+	Q_EMIT connectionStateChanged( m_connectionFlags = kError );
+}
+
+} // namespace redis
+
+} // namespace myx
+
+#endif // ifndef MYX_REDIS_BASE_INL_HPP_

src/myx/redis/base.cpp

@@ -0,0 +1,5 @@
+#ifndef MYXLIB_BUILD_LIBRARIES
+#error Define MYXLIB_BUILD_LIBRARIES to compile this file.
+#endif
+
+#include <myx/redis/base-inl.hpp>

src/myx/redis/base.hpp

@@ -0,0 +1,85 @@
+#ifndef MYX_REDIS_BASE_HPP_
+#define MYX_REDIS_BASE_HPP_
+
+#pragma once
+
+#include <myx/base/config.hpp>
+
+#include <QHash>
+#include <QObject>
+
+QT_FORWARD_DECLARE_CLASS( QTimer )
+QT_FORWARD_DECLARE_CLASS( QTcpSocket )
+
+namespace myx {
+
+namespace redis {
+
+/*!
+ * \brief Базовый класс для объектов взаимодействия с redis
+ */
+class Base : public QObject
+{
+	Q_OBJECT
+
+public:
+	/*!
+	 * \brief Состояние соединения
+	 */
+	enum ConnectionStateFlags
+	{
+		kDisconnected = 1,  //!< Соединение не установлено
+		kConnecting   = 2,  //!< Соединение устанавливается
+		kConnected    = 4,  //!< Соединение установлено
+		kRead         = 8,  //!< Есть данные для чтения
+		kWrite        = 16, //!< Есть данные для записи
+		kError        = 32, //!< Ошибка соединения
+	};
+
+	explicit Base( QObject* parent = nullptr );
+
+	/*!
+	 * \brief Сигнализирует об изменении состояния соединения
+	 * \param flags Флаги состояния
+	 */
+	Q_SIGNAL void connectionStateChanged( int flags );
+
+	bool isConnected() const;
+
+private:
+	static constexpr int k_RwTimeout { 200 };                      //!< Таймаут на операции чтения и записи
+	int                  m_connectionFlags { kDisconnected };      //!< Флаги состояния соединения
+	quint16              m_port { 6379 };                          //!< Номер порта подключения к БД
+	QString              m_host { QStringLiteral( "127.0.0.1" ) }; //!< Имя хоста подключения к БД
+	QTimer*              m_readTimer;                              //!< Таймер чтения данных
+	QTimer*              m_writeTimer;                             //!< Таймер записи данных
+
+protected:
+
+	int connectionTimeout() const;
+	int rwTimeout() const;
+
+	QString host() const;
+	void setHost( const QString& host );
+
+	quint16 port() const;
+	void setPort( quint16 port );
+
+	int connectionFlags() const;
+	void setConnectionFlags( int connectionFlags );
+
+//	int fetchLength( QByteArray& buffer, int& pos );
+	Q_SLOT void onReadyRead();
+	Q_SLOT void onBytesWritten();
+	Q_SLOT void onSocketError();
+}; // class Base
+
+} // namespace redis
+
+} // namespace myx
+
+#ifdef MYXLIB_HEADER_ONLY
+#include "base-inl.hpp"
+#endif
+
+#endif // ifndef MYX_REDIS_BASE_HPP_

src/myx/redis/containers-inl.hpp

@@ -0,0 +1,242 @@
+#ifndef MYX_REDIS_CONTAINERS_INL_HPP_
+#define MYX_REDIS_CONTAINERS_INL_HPP_
+
+#pragma once
+
+#ifndef MYXLIB_HEADER_ONLY
+#include <myx/redis/containers.hpp>
+#endif
+
+#include <QDebug>
+#include <QHash>
+
+namespace myx {
+
+namespace redis {
+
+/*!
+ * \brief Создает часть команды
+ * \param value Значение для создани части команды
+ * \return Часть команды
+ */
+inline QByteArray part( const QVariant& value )
+{
+	auto bytes = value.toByteArray();
+	return( "$" + QByteArray::number( bytes.length() ) + Separator + bytes );
+}
+
+
+/*!
+ * \brief Составляет RESP массив
+ * \param parts Части для составления массива
+ * \return RESP массив
+ */
+inline QByteArray array( const QVariantList& parts )
+{
+	QByteArrayList data;
+	std::transform( parts.cbegin(), parts.cend(), std::back_inserter( data ),
+	                []( const QVariant& iter ) { return( part( iter ) ); } );
+	data.prepend( "*" + QByteArray::number( data.size() ) );
+	return( data.join( Separator ) + Separator );
+}
+
+
+/*!
+ * \brief Разбивает буфер данных на массив частей
+ * \param buffer Входной буфер данных
+ * \param splitLength Длина разобранной части буффера
+ * \return Массив частей-строк
+ */
+inline QByteArrayList split( const QByteArray& buffer, int* splitLength )
+{
+	enum Token
+	{
+		kUndefined,
+		kFetchNumber,
+		kFetchPart,
+	};
+
+	if ( ( buffer == "$-1\r\n" ) || ( buffer == "$*0\r\n" ) || buffer.isEmpty() )
+	{
+		if ( splitLength != nullptr )
+		{
+			*splitLength = buffer.length();
+		}
+		return {};
+	}
+	if ( buffer.startsWith( "-WRONG" ) )
+	{
+		if ( splitLength != nullptr )
+		{
+			*splitLength = buffer.endsWith( "\r\n" ) ? buffer.length() : 0;
+		}
+		return {};
+	}
+	if ( splitLength != nullptr )
+	{
+		*splitLength = 0;
+	}
+	if ( !buffer.endsWith( Separator ) )
+	{
+		return {};
+	}
+	int size { 1 };
+	int pos { 0 };
+	QByteArray partLength;
+	if ( buffer[pos] == '*' )
+	{
+		pos++;
+		QByteArray text;
+		while ( QChar( buffer[pos] ).isNumber() && pos < buffer.size() )
+		{
+			text += buffer[pos++];
+		}
+		if ( pos == buffer.size() )
+		{
+			return {};
+		}
+		size = text.toInt();
+		pos += Separator.length();
+	}
+	QByteArrayList parts;
+	Token token { kUndefined };
+	while ( parts.size() < size && pos < buffer.length() )
+	{
+		switch ( token )
+		{
+		case kUndefined:
+			switch ( buffer[pos] )
+			{
+			case '$':
+				partLength.clear();
+				token = kFetchPart;
+				pos++;
+				break;
+			case ':':
+				partLength.clear();
+				token = kFetchNumber;
+				pos++;
+				break;
+			default:
+				return {};
+			}
+			break;
+		case kFetchNumber:
+			if ( QChar( buffer[pos] ).isNumber() || ( buffer[pos] == '-' ) )
+			{
+				partLength += buffer[pos];
+				pos++;
+			}
+			else
+			{
+				parts.push_back( partLength );
+				token = kUndefined;
+				pos  += Separator.length();
+			}
+			break;
+		case kFetchPart:
+		{
+			if ( QChar( buffer[pos] ).isNumber() || ( buffer[pos] == '-' ) )
+			{
+				partLength += buffer[pos];
+				pos++;
+			}
+			else
+			{
+				int length = partLength.toInt();
+				if ( length < 0 )
+				{
+					parts.push_back( "" );
+					pos += Separator.length();
+				}
+				else
+				{
+					pos += Separator.length();
+					if ( pos + length > buffer.length() )
+					{
+						return {};
+					}
+					parts.push_back( buffer.mid( pos, length ) );
+					pos += length + Separator.length();
+				}
+				token = kUndefined;
+			}
+			break;
+		}
+		} // switch
+	}
+	// может быть ситуация, когда в буфер недочитан последний \r\n,
+	// но все части были вычитаны
+	if ( ( parts.size() == size ) && ( pos <= buffer.length() ) )
+	{
+		if ( splitLength != nullptr )
+		{
+			*splitLength = pos;
+		}
+		return( parts );
+	}
+	return {};
+} // split
+
+
+/*!
+ * \brief Запаковывает сообщение для передачи через систему pubsub
+ * \param hash Таблица значений
+ * \return Запакованное сообщение
+ */
+inline QByteArray pack_hash( const QVariantHash& hash )
+{
+	QVariantList parts;
+	for ( auto iter = hash.cbegin(); iter != hash.cend(); ++iter )
+	{
+		parts << iter.key() << iter.value();
+	}
+	return( array( parts ) );
+}
+
+
+/*!
+ * \brief Распаковывает кодограмму в таблицу значений
+ * \param message Сообщение
+ * \return Таблица значений
+ */
+inline QVariantHash unpack_hash( const QByteArray& message )
+{
+	auto parts = split( message );
+	QVariantHash hash;
+	for ( int i = 1; i < parts.size(); i += 2 )
+	{
+		hash[parts[i - 1]] = parts[i];
+	}
+	return( hash );
+}
+
+
+/*!
+ * \brief Запаковывает сообщение для передачи через систему pubsub
+ * \param hash Список значений
+ * \return Запакованное сообщение
+ */
+inline QByteArray pack_list( const QVariantList& list )
+{
+	return( array( list ) );
+}
+
+
+/*!
+ * \brief Распаковывает кодограмму в список значений
+ * \param message Сообщение
+ * \return Список значений
+ */
+inline QVariantList unpack_list( const QByteArray& message )
+{
+	auto data = split( message );
+	QVariantList list;
+	std::copy( data.begin(), data.end(), std::back_inserter( list ) );
+	return( list );
+}
+
+} // namespace redis
+
+} // namespace myx
+#endif // ifndef MYX_REDIS_CONTAINERS_INL_HPP_

src/myx/redis/containers.cpp

@@ -0,0 +1,5 @@
+#ifndef MYXLIB_BUILD_LIBRARIES
+#error Define MYXLIB_BUILD_LIBRARIES to compile this file.
+#endif
+
+#include <myx/redis/containers-inl.hpp>

src/myx/redis/containers.hpp

@@ -0,0 +1,102 @@
+#ifndef MYX_REDIS_CONTAINERS_HPP_
+#define MYX_REDIS_CONTAINERS_HPP_
+
+#pragma once
+
+#include <myx/base/config.hpp>
+
+#include <QByteArrayList>
+#include <QVariantList>
+
+namespace myx {
+
+namespace redis {
+
+static const QByteArray Separator { "\r\n" }; //!< Строковый разделитель слов во фразе
+
+QByteArray array( const QVariantList& parts );
+QByteArrayList split( const QByteArray& buffer, int* splitLength = nullptr );
+// QByteArray part( const QVariant& value );
+// QByteArray packHash( const QVariantHash& hash );
+// QVariantHash unpackHash( const QByteArray& message );
+// QByteArray packList( QByteArrayList list );
+// QByteArray packList( const QVariantList& list );
+QVariantList unpack_list( const QByteArray& message );
+// template< typename T >
+// QByteArray packPair( T first, T second );
+// template< typename T1, typename T2 >
+// QByteArray packPair( T1 first, T2 second );
+// template< typename T >
+// std::pair< T, T > unpackPair( const QByteArray& message );
+// template< typename T1, typename T2 >
+// std::pair< T1, T2 > unpackPair( const QByteArray& message );
+
+
+/*!
+ * \brief Запаковывает пару значений в сообщение
+ * \param first Первое значение
+ * \param second Второе значение
+ * \return Текстовое сообщение
+ */
+template< typename T >
+QByteArray pack_pair( T first, T second )
+{
+	return( packList( { QVariant::fromValue( first ), QVariant::fromValue( second ) } ) );
+}
+
+
+/*!
+ * \brief Запаковывает пару значений в сообщение
+ * \param first Первое значение
+ * \param second Второе значение
+ * \return Текстовое сообщение
+ */
+template< typename T1, typename T2 >
+QByteArray pack_pair( T1 first, T2 second )
+{
+	return( packList( { QVariant::fromValue( first ), QVariant::fromValue( second ) } ) );
+}
+
+
+/*!
+ * \brief Распаковывает сообщение в пару однотипных значений
+ * \param pair Текст сообщения
+ * \return Пара значений
+ */
+template< typename T >
+std::pair< T, T > unpack_pair( const QByteArray& message )
+{
+	auto list = unpack_list( message );
+	if ( list.size() == 2 )
+	{
+		return { QVariant( list.first() ).value< T >(), QVariant( list.last() ).value< T >() };
+	}
+	return {};
+}
+
+
+/*!
+ * \brief Распаковывает сообщение в пару однотипных значений
+ * \param pair Текст сообщения
+ * \return Пара значений
+ */
+template< typename T1, typename T2 >
+std::pair< T1, T2 > unpack_pair( const QByteArray& message )
+{
+	auto list = unpack_list( message );
+	if ( list.size() == 2 )
+	{
+		return { QVariant( list.first() ).value< T1 >(), QVariant( list.last() ).value< T2 >() };
+	}
+	return {};
+}
+
+} // namespace redis
+
+} // namespace myx
+
+#ifdef MYXLIB_HEADER_ONLY
+#include "containers-inl.hpp"
+#endif
+
+#endif // ifndef MYX_REDIS_CONTAINERS_HPP_

src/myx/redis/pubsub-inl.hpp

@@ -0,0 +1,506 @@
+#ifndef MYX_REDIS_PUBSUB_INL_HPP_
+#define MYX_REDIS_PUBSUB_INL_HPP_
+
+#pragma once
+
+#ifndef MYXLIB_HEADER_ONLY
+#include <myx/redis/pubsub.hpp>
+#endif
+
+#include <myx/redis/containers.hpp>
+
+#include <QCoreApplication>
+#include <QDateTime>
+#include <QTcpSocket>
+#include <QThread>
+#include <QTimer>
+
+namespace myx {
+
+namespace redis {
+
+const QString PubSub::k_KeySpacePrefix { QStringLiteral( "__key%1__:" ) };
+
+/*!
+ * \brief Конструктор класса
+ * \param parent Родительский объект для m_socket
+ */
+inline PubSub::PubSub( QObject* parent ) :
+	Base             { parent },
+	m_socket         ( new QTcpSocket( this ) ),
+	m_connectionTimer( new QTimer( this ) )
+
+{
+	connect( m_socket, &QAbstractSocket::stateChanged, this, &PubSub::onSocketStateChanged );
+	connect( m_socket, &QIODevice::readyRead, this, &PubSub::onReadyRead );
+	connect( m_socket, &QIODevice::bytesWritten, this, &PubSub::onBytesWritten );
+	connect( m_socket, static_cast< void ( QAbstractSocket::* )( QAbstractSocket::SocketError ) >
+	         ( &QAbstractSocket::error ), this, [this]() {
+		qDebug() << QDateTime::currentDateTimeUtc().toString( QStringLiteral( "hh:mm:ss:zzz" ) )
+		         << QByteArray( "redis pubsub connection error" ) << m_socket->errorString();
+	} );
+	connect( m_socket, &QIODevice::readyRead, this, &PubSub::read );
+
+	m_connectionTimer->setInterval( connectionTimeout() );
+	m_connectionTimer->setSingleShot( true );
+	connect( m_connectionTimer, &QTimer::timeout, this, [this]() {
+		m_socket->connectToHost( host(), port() );
+	} );
+}
+
+
+/*!
+ * \brief Инициирует работу с БД
+ */
+inline void PubSub::start()
+{
+	Q_EMIT disconnected();
+	Q_EMIT connectionStateChanged( connectionFlags() );
+	m_socket->connectToHost( host(), port() );
+}
+
+
+/*!
+ * \brief Завершает работу с БД
+ */
+inline void PubSub::stop()
+{
+	if ( m_socket->state() == QAbstractSocket::ConnectedState )
+	{
+		m_socket->disconnectFromHost();
+		while ( m_socket->state() != QAbstractSocket::UnconnectedState )
+		{
+			QCoreApplication::processEvents();
+		}
+	}
+}
+
+
+/*!
+ * \brief Создает подписку на канал сообщений
+ * \param channel Rанал подписки
+ * \param subscriber Объект-подписчик
+ * \param method Имя метода, вызываемого при изменении получении серверного сообщения
+ */
+inline void PubSub::subscribe( const QString& channel, QObject* subscriber, const char* method )
+{
+	if ( QThread::currentThread() != thread() )
+	{
+		qWarning() << "no redis connection for caller thread";
+		return;
+	}
+	auto iter = m_subscribers.find( channel );
+	if ( iter == m_subscribers.end() )
+	{
+		if ( m_socket->state() == QAbstractSocket::ConnectedState )
+		{
+			m_socket->write( array( { "SUBSCIBE", channel } ) );
+		}
+		m_subscribers[channel].emplace_back( subscriber, method );
+	}
+	else
+	{
+		auto value = iter.value();
+		auto it    = std::find_if( value.begin(), value.end(),
+		                           [subscriber, method]( std::pair< QObject*, const char* > p ) {
+			return( p.first == subscriber && strcmp( p.second, method ) == 0 );
+		} );
+		if ( it == iter.value().end() )
+		{
+			iter.value().emplace_back( subscriber, method );
+		}
+	}
+} // PubSub::subscribe
+
+
+/*!
+ * \brief Создает подписку на канал изменения данных
+ * \param channel Rанал подписки
+ * \param subscriber Объект-подписчик
+ * \param method Имя метода, вызываемого при изменении получении серверного сообщения
+ * \param database Номер БД
+ */
+inline void PubSub::subscribe( QString channel, QObject* subscriber, const char* method, int database )
+{
+	if ( QThread::currentThread() != thread() )
+	{
+		qWarning() << "no redis connection for caller thread";
+		return;
+	}
+	auto iter = m_subscribers.find( channel );
+	if ( iter == m_subscribers.end() )
+	{
+		channel.prepend( k_KeySpacePrefix.arg( "space@" + QString::number( database ) ).toLocal8Bit() );
+		if ( m_socket->state() == QAbstractSocket::ConnectedState )
+		{
+			m_socket->write( array( { "SUBSCRIBE", channel } ) );
+		}
+		m_subscribers[channel].emplace_back( subscriber, method );
+	}
+	else
+	{
+		auto value = iter.value();
+		auto it    = std::find_if( value.begin(), value.end(),
+		                           [subscriber, method]( std::pair< QObject*, const char* > p ) {
+			return( p.first == subscriber && strcmp( p.second, method ) == 0 );
+		} );
+		if ( it == iter.value().end() )
+		{
+			iter.value().emplace_back( subscriber, method );
+		}
+	}
+} // PubSub::subscribe
+
+
+/*!
+ * \brief Создает подписку на шаблон канала сообщений
+ * \param channel Шаблон канала подписки
+ * \param subscriber Объект-подписчик
+ * \param method Имя метода, вызываемого при изменении получении серверного сообщения
+ */
+inline void PubSub::psubscribe( const QString& channel, QObject* subscriber, const char* method )
+{
+	if ( QThread::currentThread() != thread() )
+	{
+		qWarning() << "no redis connection for caller thread";
+		return;
+	}
+	auto iter = m_psubscribers.find( channel );
+	if ( iter == m_psubscribers.end() )
+	{
+		if ( m_socket->state() == QAbstractSocket::ConnectedState )
+		{
+			m_socket->write( array( { "PSUBSCRIBE", channel } ) );
+		}
+		m_psubscribers[channel].emplace_back( subscriber, method );
+	}
+	else
+	{
+		auto value = iter.value();
+		auto it    = std::find_if( value.begin(), value.end(),
+		                           [subscriber, method]( std::pair< QObject*, const char* > p ) {
+			return( p.first == subscriber && strcmp( p.second, method ) == 0 );
+		} );
+		if ( it == iter.value().end() )
+		{
+			iter.value().emplace_back( subscriber, method );
+		}
+	}
+} // PubSub::psubscribe
+
+
+/*!
+ * \brief Создает подписку на шаблон канала изменения данных
+ * \param channel Шаблон канала подписки
+ * \param subscriber Объект-подписчик
+ * \param method Имя метода, вызываемого при изменении получении серверного сообщения
+ * \param database Номер БД (-1 для всех)
+ */
+inline void PubSub::psubscribe( QString channel, QObject* subscriber, const char* method, int database )
+{
+	if ( QThread::currentThread() != thread() )
+	{
+		qWarning() << "no redis connection for caller thread";
+		return;
+	}
+	auto iter = m_psubscribers.find( channel );
+	if ( iter == m_psubscribers.end() )
+	{
+		if ( database == -1 )
+		{
+			channel.prepend( k_KeySpacePrefix.arg( QStringLiteral( "*" ) ).toLocal8Bit() );
+		}
+		else
+		{
+			channel.prepend( k_KeySpacePrefix.arg( "space@" + QString::number( database ) ).toLocal8Bit() );
+		}
+		if ( m_socket->state() == QAbstractSocket::ConnectedState )
+		{
+			m_socket->write( array( { "PSUBSCRIBE", channel } ) );
+		}
+		m_psubscribers[channel].emplace_back( subscriber, method );
+	}
+	else
+	{
+		auto value = iter.value();
+		auto it    = std::find_if( value.begin(), value.end(),
+		                           [subscriber, method]( std::pair< QObject*, const char* > p ) {
+			return( p.first == subscriber && strcmp( p.second, method ) == 0 );
+		} );
+		if ( it == iter.value().end() )
+		{
+			iter.value().emplace_back( subscriber, method );
+		}
+	}
+} // PubSub::psubscribe
+
+
+/*!
+ * \brief Отписывает получателя от канала
+ * \param channel Имя канала
+ * \param subscriber Подписчик на канал
+ * \param method Метод обработки данных
+ * \param keyEvents Признак подписки на изменение данных
+ */
+inline void PubSub::unsubscribe( const QString& channel, QObject* subscriber, const char* method )
+{
+	if ( QThread::currentThread() != thread() )
+	{
+		qWarning() << "no redis connection for caller thread";
+		return;
+	}
+	auto iter = m_subscribers.find( channel );
+	if ( iter != m_subscribers.end() )
+	{
+		auto ctx = std::find_if( iter->begin(), iter->end(),
+		                         [subscriber, method]( std::pair< QObject*, const char* > value ) {
+			return( subscriber == value.first && strcmp( method, value.second ) == 0 );
+		} );
+		if ( ctx != iter->end() )
+		{
+			iter->erase( ctx );
+		}
+		if ( iter->empty() )
+		{
+			if ( m_socket->state() == QAbstractSocket::ConnectedState )
+			{
+				m_socket->write( array( { "UNSUBSCRIBE", channel } ) );
+			}
+			m_subscribers.remove( channel );
+		}
+	}
+} // PubSub::unsubscribe
+
+
+/*!
+ * \brief Отписывает получателя от канала
+ * \param channel Имя канала
+ * \param subscriber Подписчик на канал
+ * \param method Метод обработки данных
+ * \param keyEvents Признак подписки на изменение данных
+ */
+inline void PubSub::unsubscribe( QString channel, QObject* subscriber, const char* method, int database )
+{
+	if ( QThread::currentThread() != thread() )
+	{
+		qWarning() << "no redis connection for caller thread";
+		return;
+	}
+	if ( database == -1 )
+	{
+		channel.prepend( k_KeySpacePrefix.arg( QStringLiteral( "*" ) ).toLocal8Bit() );
+	}
+	else
+	{
+		channel.prepend( k_KeySpacePrefix.arg( "space@" + QString::number( database ) ).toLocal8Bit() );
+	}
+	auto iter = m_subscribers.find( channel );
+	if ( iter != m_subscribers.end() )
+	{
+		auto ctx = std::find_if( iter->begin(), iter->end(),
+		                         [subscriber, method]( std::pair< QObject*, const char* > value ) {
+			return( subscriber == value.first && strcmp( method, value.second ) == 0 );
+		} );
+		if ( ctx != iter->end() )
+		{
+			iter->erase( ctx );
+		}
+		if ( iter->empty() )
+		{
+			if ( m_socket->state() == QAbstractSocket::ConnectedState )
+			{
+				m_socket->write( array( { "UNSUBSCRIBE", channel } ) );
+			}
+			m_subscribers.remove( channel );
+		}
+	}
+} // PubSub::unsubscribe
+
+
+/*!
+ * \brief Отписывает получателя от шаблона канала
+ * \param channel Шаблон канала
+ * \param subscriber Подписчик на канал
+ * \param method Метод обработки данных
+ * \param keyEvents Признак подписки на изменение данных
+ */
+inline void PubSub::punsubscribe( const QString& channel, QObject* subscriber,
+                                  const char* method )
+{
+	if ( QThread::currentThread() != thread() )
+	{
+		qWarning() << "no redis connection for caller thread";
+		return;
+	}
+	auto iter = m_psubscribers.find( channel );
+	if ( iter != m_psubscribers.end() )
+	{
+		auto ctx = std::find_if( iter->begin(), iter->end(),
+		                         [subscriber, method]( std::pair< QObject*, const char* > value ) {
+			return( subscriber == value.first && strcmp( method, value.second ) == 0 );
+		} );
+		if ( ctx != iter->end() )
+		{
+			iter->erase( ctx );
+		}
+		if ( iter->empty() )
+		{
+			if ( m_socket->state() == QAbstractSocket::ConnectedState )
+			{
+				m_socket->write( array( { "PUNSUBSCRIBE", channel } ) );
+			}
+			m_psubscribers.remove( channel );
+		}
+	}
+} // PubSub::punsubscribe
+
+
+/*!
+ * \brief Отписывает получателя от шаблона канала
+ * \param channel Шаблон канала
+ * \param subscriber Подписчик на канал
+ * \param method Метод обработки данных
+ * \param keyEvents Признак подписки на изменение данных
+ */
+inline void PubSub::punsubscribe( QString channel, QObject* subscriber,
+                                  const char* method, int database )
+{
+	if ( QThread::currentThread() != thread() )
+	{
+		qWarning() << "no redis connection for caller thread";
+		return;
+	}
+	if ( database == -1 )
+	{
+		channel.prepend( k_KeySpacePrefix.arg( QStringLiteral( "*" ) ).toLocal8Bit() );
+	}
+	else
+	{
+		channel.prepend( k_KeySpacePrefix.arg( "space@" + QString::number( database ) ).toLocal8Bit() );
+	}
+	auto iter = m_psubscribers.find( channel );
+	if ( iter != m_psubscribers.end() )
+	{
+		auto ctx = std::find_if( iter->begin(), iter->end(),
+		                         [subscriber, method]( std::pair< QObject*, const char* > value ) {
+			return( subscriber == value.first && strcmp( method, value.second ) == 0 );
+		} );
+		if ( ctx != iter->end() )
+		{
+			iter->erase( ctx );
+		}
+		if ( iter->empty() )
+		{
+			if ( m_socket->state() == QAbstractSocket::ConnectedState )
+			{
+				m_socket->write( array( { "PUNSUBSCRIBE", channel } ) );
+			}
+			m_psubscribers.remove( channel );
+		}
+	}
+} // PubSub::punsubscribe
+
+
+/*!
+ * \brief Разбирает поступившие от сервера уведомления и вызывает требуемые
+ * метаметоды обработки
+ */
+inline void PubSub::read()
+{
+	int splitLength = 0;
+	while ( m_socket->bytesAvailable() > 0 )
+	{
+		m_buffer += m_socket->readAll();
+		while ( !m_buffer.isEmpty() )
+		{
+			auto parts = split( m_buffer, &splitLength );
+			if ( parts.empty() )
+			{
+				break;
+			}
+			if ( ( parts[0] == "pmessage" ) && ( parts.size() == 4 ) )
+			{
+				auto iter = m_psubscribers.find( parts[1] );
+				if ( iter != m_psubscribers.end() )
+				{
+					for ( const auto& i: iter.value() )
+					{
+						QMetaObject::invokeMethod( i.first, i.second,
+						                           Q_ARG( QString, parts[2] ),
+						                           Q_ARG( QByteArray, parts[3] ) );
+					}
+				}
+			}
+			else
+			{
+				if ( ( parts[0] == "message" ) && ( parts.size() == 3 ) )
+				{
+					auto iter = m_subscribers.find( parts[1] );
+					if ( iter != m_subscribers.end() )
+					{
+						for ( const auto& i: iter.value() )
+						{
+							QMetaObject::invokeMethod( i.first, i.second,
+							                           Q_ARG( QString, parts[1] ),
+							                           Q_ARG( QByteArray, parts[2] ) );
+						}
+					}
+				}
+			}
+			m_buffer.remove( 0, splitLength );
+		}
+	}
+} // PubSub::read
+
+
+/*!
+ * \brief Обработка изменения состояния сокета
+ */
+inline void PubSub::onSocketStateChanged()
+{
+	switch ( m_socket->state() )
+	{
+	case QAbstractSocket::ConnectedState:
+		for ( auto iter = m_psubscribers.cbegin(); iter != m_psubscribers.cend(); ++iter )
+		{
+			m_socket->write( array( { "PSUBSCRIBE", iter.key() } ) );
+		}
+		for ( auto iter = m_subscribers.cbegin(); iter != m_subscribers.cend(); ++iter )
+		{
+			m_socket->write( array( { "SUBSCRIBE", iter.key() } ) );
+		}
+		setConnectionFlags( kConnected );
+		Q_EMIT connectionStateChanged( connectionFlags() );
+		Q_EMIT connected();
+		break;
+	case QAbstractSocket::UnconnectedState:
+		m_buffer.clear();
+		m_connectionTimer->start();
+		if ( !isConnected() )
+		{
+			Q_EMIT disconnected();
+		}
+		setConnectionFlags( kDisconnected );
+		Q_EMIT connectionStateChanged( connectionFlags() );
+		break;
+	default:
+		break;
+	} // switch
+} // PubSub::onSocketStateChanged
+
+
+/*!
+ * \brief Возвращает сокет
+ * \return Сокет
+ */
+inline QTcpSocket* PubSub::socket() const
+{
+	return( m_socket );
+}
+
+} // namespace redis
+
+} // namespace myx
+
+
+#endif // ifndef MYX_REDIS_PUBSUB_INL_HPP_

src/myx/redis/pubsub.cpp

@@ -0,0 +1,5 @@
+#ifndef MYXLIB_BUILD_LIBRARIES
+#error Define MYXLIB_BUILD_LIBRARIES to compile this file.
+#endif
+
+#include <myx/redis/pubsub-inl.hpp>

src/myx/redis/pubsub.hpp

@@ -0,0 +1,76 @@
+#ifndef MYX_REDIS_PUBSUB_HPP_
+#define MYX_REDIS_PUBSUB_HPP_
+
+#pragma once
+
+#include <myx/redis/base.hpp>
+
+#include <QObject>
+
+QT_FORWARD_DECLARE_CLASS( QTcpSocket )
+
+namespace myx {
+
+namespace redis {
+
+/*!
+ * \brief УПравляет подпиской и публикацией сообщений
+ */
+
+class PubSub : public Base
+{
+	Q_OBJECT
+
+public:
+	explicit PubSub( QObject* parent = nullptr );
+	QTcpSocket* socket() const;
+	Q_SLOT void start();
+	Q_SLOT void stop();
+	Q_INVOKABLE void subscribe( const QString& channel, QObject* subscriber, const char* method );
+	Q_INVOKABLE void subscribe( QString channel, QObject* subscriber,
+	                            const char* method, int database );
+	Q_INVOKABLE void psubscribe( const QString& channel, QObject* subscriber, const char* method );
+	Q_INVOKABLE void psubscribe( QString channel, QObject* subscriber,
+	                             const char* method, int database );
+	Q_INVOKABLE void unsubscribe( const QString& channel, QObject* subscriber, const char* method );
+	Q_INVOKABLE void unsubscribe( QString channel, QObject* subscriber,
+	                              const char* method, int database );
+	Q_INVOKABLE void punsubscribe( const QString& channel, QObject* subscriber, const char* method );
+	Q_INVOKABLE void punsubscribe( QString channel, QObject* subscriber,
+	                               const char* method, int database );
+	/*!
+	 * \brief Сигнализирует об установке соединения
+	 */
+	Q_SIGNAL void connected();
+	/*!
+	 * \brief Сигнализирует о разрыве соединения
+	 */
+	Q_SIGNAL void disconnected();
+
+private:
+	using Sub = QHash< QString, std::vector< std::pair< QObject*, const char* > > >;
+
+	static const QString k_KeySpacePrefix;             //!< Приставка для наблюдения за данными
+	QTcpSocket*          m_socket;
+	QTimer*              m_connectionTimer;            //!< Таймер установления сетевого соединения
+	static constexpr int k_ConnectionTimeout { 2000 }; //!< Таймаут на сетевые операции
+	/*!
+	 * \brief Таблица подписчиков
+	 */
+	QByteArray m_buffer;                               //!< Буффер прочитанных данных
+	Sub        m_subscribers;                          //!< Обработчики сообщений по каналу подписки
+	Sub        m_psubscribers;                         //!< Обработчики сообщений по шаблону канала подписки
+
+	Q_SLOT void read();
+	Q_SLOT void onSocketStateChanged();
+}; // class PubSub
+
+} // namespace redis
+
+} // namespace myx
+
+#ifdef MYXLIB_HEADER_ONLY
+#include "pubsub-inl.hpp"
+#endif
+
+#endif // ifndef MYX_REDIS_PUBSUB_HPP_