Compare commits
No commits in common. "5936bf96f00e2e4941620d348c53903a81d62ea2" and "e6253e777f72be323381b92a0bd3701d418d1826" have entirely different histories.
5936bf96f0
...
e6253e777f
@ -223,7 +223,7 @@ CheckOptions:
|
||||
- key: readability-identifier-naming.ProtectedMemberCase
|
||||
value: camelBack
|
||||
- key: readability-identifier-naming.ProtectedMemberPrefix
|
||||
value: 'm_'
|
||||
value: ''
|
||||
- key: readability-identifier-naming.ProtectedMemberSuffix
|
||||
value: ''
|
||||
- key: readability-identifier-naming.ProtectedMethodCase
|
||||
|
||||
@ -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()
|
||||
|
||||
# Документация
|
||||
|
||||
@ -1 +1 @@
|
||||
Subproject commit 2ea38c32c8d6aa7ccdbdf84cff1d0b0d50cc2fb3
|
||||
Subproject commit 1eaded9031726f539cd5fd4a44d874d7bd8d458d
|
||||
@ -1,72 +0,0 @@
|
||||
# Название основной цели и имя библиотеки в текущем каталоге
|
||||
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")
|
||||
@ -4,20 +4,24 @@ set(TRGT redis)
|
||||
# cmake-format: off
|
||||
# Список файлов исходных текстов
|
||||
set(TRGT_cpp
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/base.cpp
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/pubsub.cpp
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/containers.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}/base.hpp
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/pubsub.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}/base-inl.hpp
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/containers.hpp
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/containers-inl.hpp
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/pubsub-inl.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
|
||||
|
||||
@ -1,173 +0,0 @@
|
||||
#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_
|
||||
@ -1,5 +0,0 @@
|
||||
#ifndef MYXLIB_BUILD_LIBRARIES
|
||||
#error Define MYXLIB_BUILD_LIBRARIES to compile this file.
|
||||
#endif
|
||||
|
||||
#include <myx/redis/base-inl.hpp>
|
||||
@ -1,85 +0,0 @@
|
||||
#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_
|
||||
@ -1,242 +0,0 @@
|
||||
#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_
|
||||
@ -1,5 +0,0 @@
|
||||
#ifndef MYXLIB_BUILD_LIBRARIES
|
||||
#error Define MYXLIB_BUILD_LIBRARIES to compile this file.
|
||||
#endif
|
||||
|
||||
#include <myx/redis/containers-inl.hpp>
|
||||
@ -1,102 +0,0 @@
|
||||
#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_
|
||||
@ -1,506 +0,0 @@
|
||||
#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_
|
||||
@ -1,5 +0,0 @@
|
||||
#ifndef MYXLIB_BUILD_LIBRARIES
|
||||
#error Define MYXLIB_BUILD_LIBRARIES to compile this file.
|
||||
#endif
|
||||
|
||||
#include <myx/redis/pubsub-inl.hpp>
|
||||
@ -1,76 +0,0 @@
|
||||
#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_
|
||||
Loading…
Reference in New Issue
Block a user