Skip to content

dataproxy
dataproxy

colorEagleStdio / dataproxy

通知 9
Star 6
Fork 0
dataproxy
dataproxy

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

提交 538ce8fd 编写于 作者: 丁劲犇's avatar 丁劲犇 😸
浏览文件
操作

改变了基础引擎，使得其可以记住主动连接的东西

上级 49ae07bc
隐藏空白更改
内联 并排
Showing with 64 addition and 43 deletion
network/zp_net_threadpool.cpp
浏览文件 @ 538ce8fd
......@@ -420,9 +420,11 @@ namespace ZPNetwork{
* @param address the address to connect to
* @param nPort port to connect to
* @param bSSLConn if true, SSL connections will be used
* @param extraData Extra data for users to remember extra imformation for this connection.this message will be
* given back when on_connect is called.
* @return bool
*/
bool zp_net_Engine::connectTo (const QHostAddress & address , quint16 nPort,bool bSSLConn)
bool zp_net_Engine::connectTo (const QHostAddress & address , quint16 nPort,bool bSSLConn, quint64 extraData)
{
bool res= false;
//m_mutex_trans.lock();
......@@ -453,7 +455,7 @@ namespace ZPNetwork{
if (nMinIdx>=0 && nMinIdx<nsz)
{
res = true;
emit evt_FireConnection(m_vec_NetTransThreads[nMinIdx],address,nPort);
emit evt_FireConnection(m_vec_NetTransThreads[nMinIdx],address,nPort,extraData);
}
else
{
......
network/zp_net_threadpool.h
浏览文件 @ 538ce8fd
......@@ -75,24 +75,24 @@ namespace ZPNetwork{
void evt_Message(QObject * pSource,QString );
//The socket error message
void evt_SocketError(QObject * senderSock ,QAbstractSocket::SocketError socketError);
void evt_SocketError(QObject * senderSock ,QAbstractSocket::SocketError socketError, quint64 extraData);
//this event indicates new client connected.
void evt_NewClientConnected(QObject * /*clientHandle*/);
void evt_NewClientConnected(QObject * /*clientHandle*/, quint64 extraData);
//SSL Connections OK
void evt_ClientEncrypted(QObject * client);
void evt_ClientEncrypted(QObject * client, quint64 extraData);
//this event indicates a client disconnected.
void evt_ClientDisconnected(QObject * /*clientHandle*/);
void evt_ClientDisconnected(QObject * /*clientHandle*/, quint64 extraData);
//some data arrival
void evt_Data_recieved(QObject * /*clientHandle*/,QByteArray /*datablock*/ );
void evt_Data_recieved(QObject * /*clientHandle*/,QByteArray /*datablock*/ , quint64 extraData);
//a block of data has been successfuly sent
void evt_Data_transferred(QObject * /*clientHandle*/,qint64 /*bytes sent*/);
void evt_Data_transferred(QObject * /*clientHandle*/,qint64 /*bytes sent*/, quint64 extraData);
//Internal Message for ctrl.------------------------------------
//Listen Control
void startListen(QString id);
void stopListen(QString id);
void evt_EstablishConnection(QObject * threadid,qintptr socketDescriptor);
void evt_FireConnection(QObject * threadid,const QHostAddress & hostAddr, quint16 port);
void evt_FireConnection(QObject * threadid,const QHostAddress & hostAddr, quint16 port, quint64 extraData);
//Trans Control,for intenal thread usage
void evt_SendDataToClient(QObject * objClient,QByteArray dtarray);
void evt_KickClient(QObject *);
......@@ -109,7 +109,7 @@ namespace ZPNetwork{
void KickClients(QObject * object);
//Possive Connection Methods
bool connectTo (const QHostAddress & address , quint16 nPort,bool bSSLConn);
bool connectTo (const QHostAddress & address , quint16 nPort,bool bSSLConn, quint64 extraData);
};
}
......
network/zp_nettransthread.cpp
浏览文件 @ 538ce8fd
......@@ -130,6 +130,7 @@ namespace ZPNetwork{
while (m_rabish_can.size()>=RUBBISH_CAN_SIZE)
{
if (m_extraData.contains(m_rabish_can.first())) m_extraData.remove(m_rabish_can.first());
m_rabish_can.first()->deleteLater();
m_set_rabish.remove(m_rabish_can.first());
m_rabish_can.pop_front();
......@@ -161,6 +162,10 @@ namespace ZPNetwork{
#endif
if (sock_client)
{
quint64 extraData = 0;
if (m_extraData.contains(sock_client))
extraData = m_extraData[sock_client];
//Initial content
if (true ==sock_client->setSocketDescriptor(socketDescriptor))
{
......@@ -186,7 +191,7 @@ namespace ZPNetwork{
#endif
qDebug()<<sock_client->peerAddress().toString()<<
sock_client->peerPort() <<tr("(%1)..Accepted.").arg((quint64)sock_client);
emit evt_NewClientConnected(sock_client);
emit evt_NewClientConnected(sock_client,extraData);
//emit evt_Message(sock_client,"Info>" + QString(tr("Client Accepted.")));
}
else
......@@ -202,7 +207,7 @@ namespace ZPNetwork{
* @param addr address to which the socket should be connected.
* @param port port to which the socket should be connected.
*/
void zp_netTransThread::startConnection(QObject * threadid,const QHostAddress & addr, quint16 port)
void zp_netTransThread::startConnection(QObject * threadid,const QHostAddress & addr, quint16 port, quint64 extraData)
{
if (threadid!=this)
return;
......@@ -217,6 +222,7 @@ namespace ZPNetwork{
#endif
if (sock_client)
{
m_extraData[sock_client] = extraData;
#if (ZP_WANTSSL!=0)
if (m_bSSLConnection==true)
{
......@@ -270,7 +276,9 @@ namespace ZPNetwork{
void zp_netTransThread::on_connected()
{
QTcpSocket * pSock = qobject_cast<QTcpSocket*>(sender());
emit evt_NewClientConnected(pSock);
quint64 extraData = 0;
if (m_extraData.contains(pSock)) extraData = m_extraData[pSock];
emit evt_NewClientConnected(pSock,extraData);
emit evt_Message(pSock,"Info>" + QString(tr("Client connected.")));
qDebug()<<pSock->peerAddress().toString()<<
pSock->peerPort() <<tr("(%1)..connected.").arg((quint64)pSock);
......@@ -279,7 +287,9 @@ namespace ZPNetwork{
void zp_netTransThread::on_encrypted()
{
QTcpSocket * pSock = qobject_cast<QTcpSocket*>(sender());
emit evt_ClientEncrypted(pSock);
quint64 extraData = 0;
if (m_extraData.contains(pSock)) extraData = m_extraData[pSock];
emit evt_ClientEncrypted(pSock,extraData);
//emit evt_Message(pSock,"Info>" + QString(tr("Client Encrypted.")));
qDebug()<<pSock->peerAddress().toString()<<
pSock->peerPort() <<tr("(%1)..Encrypted.").arg((quint64)pSock);
......@@ -290,6 +300,9 @@ namespace ZPNetwork{
QTcpSocket * pSock = qobject_cast<QTcpSocket*>(sender());
if (pSock)
{
quint64 extraData = 0;
if (m_extraData.contains(pSock)) extraData = m_extraData[pSock];
#if (ZP_WANTSSL!=0)
if (m_bSSLConnection)
{
......@@ -310,7 +323,7 @@ namespace ZPNetwork{
m_clientList.remove(pSock);
m_mutex_protect.unlock();
//pSock->abort();
emit evt_ClientDisconnected(pSock);
emit evt_ClientDisconnected(pSock,extraData);
emit evt_Message(pSock,"Info>" + QString(tr("Client Closed.")));
qDebug()<<tr("(%1)..Closed.").arg((quint64)pSock);
push_to_rabish_can(pSock);
......@@ -321,13 +334,16 @@ namespace ZPNetwork{
QTcpSocket * pSock = qobject_cast<QTcpSocket*>(sender());
if (pSock)
{
quint64 extraData = 0;
if (m_extraData.contains(pSock)) extraData = m_extraData[pSock];
QByteArray array = pSock->readAll();
int sz = array.size();
g_mutex_sta.lock();
g_bytesRecieved +=sz;
g_secRecieved += sz;
g_mutex_sta.unlock();
emit evt_Data_recieved(pSock,array);
emit evt_Data_recieved(pSock,array,extraData);
}
}
/**
......@@ -346,7 +362,10 @@ namespace ZPNetwork{
QTcpSocket * pSock = qobject_cast<QTcpSocket*>(sender());
if (pSock)
{
emit evt_Data_transferred(pSock,wsended);
quint64 extraData = 0;
if (m_extraData.contains(pSock)) extraData = m_extraData[pSock];
emit evt_Data_transferred(pSock,wsended,extraData);
QList<QByteArray> & list_sock_data = m_buffer_sending[pSock];
QList<qint64> & list_offset = m_buffer_sending_offset[pSock];
while (list_sock_data.empty()==false)
......@@ -372,9 +391,12 @@ namespace ZPNetwork{
QTcpSocket * pSock = qobject_cast<QTcpSocket*>(sender());
if (pSock)
{
quint64 extraData = 0;
if (m_extraData.contains(pSock)) extraData = m_extraData[pSock];
qDebug()<<pSock->peerAddress().toString()<<
pSock->peerPort() <<tr("(%1)..Error :%2.").arg((quint64)pSock).arg(pSock->errorString());
emit evt_SocketError(pSock,socketError);
emit evt_SocketError(pSock,socketError,extraData);
#if (ZP_WANTSSL!=0)
if (m_bSSLConnection)
{
......@@ -394,7 +416,7 @@ namespace ZPNetwork{
m_clientList.remove(pSock);
m_mutex_protect.unlock();
pSock->abort();
emit evt_ClientDisconnected(pSock);
emit evt_ClientDisconnected(pSock,extraData);
emit evt_Message(pSock,"Info>" + QString(tr("Client Error, Closed.")));
//pSock->disconnectFromHost();
push_to_rabish_can(pSock);
......
network/zp_nettransthread.h
浏览文件 @ 538ce8fd
......@@ -53,12 +53,14 @@ namespace ZPNetwork{
QList<QObject *> m_rabish_can;
QSet<QObject *> m_set_rabish;
QMutex m_mutex_rabish_can;
//Extra Data
QHash<QObject *, quint64> m_extraData;
void push_to_rabish_can(QObject * deletedobj);
public slots:
//This slot dealing with multi-thread client socket accept.
void incomingConnection(QObject * threadid,qintptr socketDescriptor);
//This slot dealing with possive connecting-to method.
void startConnection(QObject * threadid,const QHostAddress & addr, quint16 port);
void startConnection(QObject * threadid,const QHostAddress & addr, quint16 port, quint64 extraData);
//sending dtarray to objClient. dtarray will be pushed into m_buffer_sending
void SendDataToClient(QObject * objClient,QByteArray dtarray);
//Set terminate mark, the thread will quit after last client quit.
......@@ -82,12 +84,12 @@ namespace ZPNetwork{
void on_encrypted();
signals:
void evt_Message(QObject * psource,QString );
void evt_SocketError(QObject * senderSock ,QAbstractSocket::SocketError socketError);
void evt_NewClientConnected(QObject * client);
void evt_ClientEncrypted(QObject * client);
void evt_ClientDisconnected(QObject * client);
void evt_Data_recieved(QObject * ,QByteArray );
void evt_Data_transferred(QObject * client,qint64);
void evt_SocketError(QObject * senderSock ,QAbstractSocket::SocketError socketError, quint64 extraData);
void evt_NewClientConnected(QObject * client, quint64 extraData);
void evt_ClientEncrypted(QObject * client, quint64 extraData);
void evt_ClientDisconnected(QObject * client, quint64 extraData);
void evt_Data_recieved(QObject * ,QByteArray , quint64 extraData);
void evt_Data_transferred(QObject * client,qint64, quint64 extraData);
};
}
#endif // ZP_NETTRANSTHREAD_H
proxyobject.cpp
浏览文件 @ 538ce8fd
......@@ -81,29 +81,28 @@ void ProxyObject::initEngine()
}
//The socket error message
void ProxyObject::slot_SocketError(QObject * senderSock ,QAbstractSocket::SocketError socketError)
void ProxyObject::slot_SocketError(QObject * senderSock ,QAbstractSocket::SocketError socketError,quint64)
{
QString msg = tr(",Source=%1, SockError = %2").arg((quint64)senderSock).arg((quint64)socketError);
qWarning()<<msg;
}
//this event indicates new client connected.
void ProxyObject::slot_NewClientConnected(QObject * clientHandle)
void ProxyObject::slot_NewClientConnected(QObject * clientHandle,quint64 extraData)
{
QTcpSocket * sock = qobject_cast<QTcpSocket *> (clientHandle);
if (sock)
{
QString pn = sock->peerName();
if (pn.length())
if (extraData)
{
if (m_para_IPLocalPort.contains(pn))
{
qDebug()<<"Outer side " << pn<<":"<<sock->peerPort()<<",Local Port="<<sock->localPort()<<" Connected";
int nLocalPort = m_para_IPLocalPort[pn];
if (m_pendingInners[nLocalPort].size())
QObject * innerClient = reinterpret_cast<QObject *> (extraData);
if (innerClient)
{
QObject * innerClient = m_pendingInners[nLocalPort].first();
m_pendingInners[nLocalPort].pop_front();
m_hash_Inner2Outer[innerClient] = clientHandle;
m_hash_Outer2Inner[clientHandle] = innerClient;
if (penging_data.contains(innerClient))
......@@ -144,8 +143,7 @@ void ProxyObject::slot_NewClientConnected(QObject * clientHandle)
if (m_para_OuterPort.contains(localPort))
{
qDebug()<<"Inner side "<<sock->peerAddress().toString()<<":"<<sock->peerPort()<<",Local Port="<<sock->localPort()<<" Connected";
m_pendingInners[localPort].push_back(clientHandle);
engine->connectTo(QHostAddress(m_para_OuterAddress[localPort]),m_para_OuterPort[localPort],false);
engine->connectTo(QHostAddress(m_para_OuterAddress[localPort]),m_para_OuterPort[localPort],false,reinterpret_cast<quint64>(sock));
}
else
{
......@@ -158,17 +156,15 @@ void ProxyObject::slot_NewClientConnected(QObject * clientHandle)
}
//this event indicates a client disconnected.
void ProxyObject::slot_ClientDisconnected(QObject * clientHandle)
void ProxyObject::slot_ClientDisconnected(QObject * clientHandle,quint64)
{
penging_data.remove(clientHandle);
m_hash_Inner2Outer.remove(clientHandle);
m_hash_Outer2Inner.remove(clientHandle);
foreach (int k , m_pendingInners.keys())
m_pendingInners[k].removeAll(clientHandle);
}
//some data arrival
void ProxyObject::slot_Data_recieved(QObject * clientHandle,QByteArray datablock )
void ProxyObject::slot_Data_recieved(QObject * clientHandle,QByteArray datablock,quint64 )
{
if (m_hash_Inner2Outer.contains(clientHandle))
engine->SendDataToClient(m_hash_Inner2Outer[clientHandle],datablock);
......
proxyobject.h
浏览文件 @ 538ce8fd
......@@ -24,20 +24,19 @@ private:
private:
int m_nTimerRefresh = -1;
ZPNetwork::zp_net_Engine * engine;
QHash<int, QList<QObject *> > m_pendingInners;
QHash<QObject *, QObject *> m_hash_Inner2Outer;
QHash<QObject *, QObject *> m_hash_Outer2Inner;
QHash<QObject *, QList< QByteArray > > penging_data;
public slots:
void slot_Message(QObject * pSource,QString );
//The socket error message
void slot_SocketError(QObject * senderSock ,QAbstractSocket::SocketError socketError);
void slot_SocketError(QObject * senderSock ,QAbstractSocket::SocketError socketError,quint64);
//this event indicates new client connected.
void slot_NewClientConnected(QObject * /*clientHandle*/);
void slot_NewClientConnected(QObject * /*clientHandle*/,quint64);
//this event indicates a client disconnected.
void slot_ClientDisconnected(QObject * /*clientHandle*/);
void slot_ClientDisconnected(QObject * /*clientHandle*/,quint64);
//some data arrival
void slot_Data_recieved(QObject * /*clientHandle*/,QByteArray /*datablock*/ );
void slot_Data_recieved(QObject * /*clientHandle*/,QByteArray /*datablock*/ ,quint64);
};
#endif // PROXYOBJECT_H
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或