MXS-1752: Remove EOF packet generation
The creation of the EOF packet is not needed as the last packet of a result set is always guaranteed to be of the correct type. This also allows non-resultsets to be correctly processed as the internal packet number will be at 0 when the last result arrives. Cleaned up some of the function names and changed the signatures to be better suited for their use-cases. Use angle bracket includes, combine some of the more unwieldly conditionals into functions, added more comments.
This commit is contained in:
@ -10,11 +10,11 @@
|
|||||||
* of this software will be governed by version 2 or later of the General
|
* of this software will be governed by version 2 or later of the General
|
||||||
* Public License.
|
* Public License.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
#include "cat.hh"
|
#include "cat.hh"
|
||||||
#include "catsession.hh"
|
#include "catsession.hh"
|
||||||
#include "maxscale/protocol/mysql.h"
|
|
||||||
#include "maxscale/modutil.h"
|
#include <maxscale/protocol/mysql.h>
|
||||||
|
#include <maxscale/modutil.h>
|
||||||
|
|
||||||
using namespace maxscale;
|
using namespace maxscale;
|
||||||
|
|
||||||
@ -22,7 +22,9 @@ CatSession::CatSession(MXS_SESSION* session, Cat* router, SRWBackendList& backen
|
|||||||
RouterSession(session),
|
RouterSession(session),
|
||||||
m_session(session),
|
m_session(session),
|
||||||
m_backends(backends),
|
m_backends(backends),
|
||||||
m_completed(0)
|
m_completed(0),
|
||||||
|
m_packet_num(0),
|
||||||
|
m_query(NULL)
|
||||||
{
|
{
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -34,13 +36,15 @@ void CatSession::close()
|
|||||||
{
|
{
|
||||||
}
|
}
|
||||||
|
|
||||||
void CatSession::skip_unused()
|
bool CatSession::next_backend()
|
||||||
{
|
{
|
||||||
// Skip unused backends
|
// Skip unused backends
|
||||||
while (m_current != m_backends.end() && !(*m_current)->in_use())
|
while (m_current != m_backends.end() && !(*m_current)->in_use())
|
||||||
{
|
{
|
||||||
m_current++;
|
m_current++;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
return m_current != m_backends.end();
|
||||||
}
|
}
|
||||||
|
|
||||||
int32_t CatSession::routeQuery(GWBUF* pPacket)
|
int32_t CatSession::routeQuery(GWBUF* pPacket)
|
||||||
@ -52,10 +56,7 @@ int32_t CatSession::routeQuery(GWBUF* pPacket)
|
|||||||
m_query = pPacket;
|
m_query = pPacket;
|
||||||
m_current = m_backends.begin();
|
m_current = m_backends.begin();
|
||||||
|
|
||||||
// If the first backend is not in use, find one that is
|
if (next_backend())
|
||||||
skip_unused();
|
|
||||||
|
|
||||||
if (m_current != m_backends.end())
|
|
||||||
{
|
{
|
||||||
// We have a backend, write the query only to this one. It will be
|
// We have a backend, write the query only to this one. It will be
|
||||||
// propagated onwards in clientReply.
|
// propagated onwards in clientReply.
|
||||||
@ -71,39 +72,23 @@ void CatSession::clientReply(GWBUF* pPacket, DCB* pDcb)
|
|||||||
auto backend = *m_current;
|
auto backend = *m_current;
|
||||||
ss_dassert(backend->dcb() == pDcb);
|
ss_dassert(backend->dcb() == pDcb);
|
||||||
bool send = false;
|
bool send = false;
|
||||||
bool propagate = true;
|
|
||||||
|
|
||||||
if (m_completed == 0 && backend->get_reply_state() == REPLY_STATE_START &&
|
|
||||||
!mxs_mysql_is_result_set(pPacket))
|
|
||||||
{
|
|
||||||
propagate = false;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (backend->reply_is_complete(pPacket))
|
if (backend->reply_is_complete(pPacket))
|
||||||
{
|
{
|
||||||
backend->ack_write();
|
backend->ack_write();
|
||||||
m_completed++;
|
m_completed++;
|
||||||
m_current++;
|
m_current++;
|
||||||
skip_unused();
|
|
||||||
|
|
||||||
if (m_current == m_backends.end())
|
if (!next_backend())
|
||||||
{
|
{
|
||||||
uint8_t eof_packet[] = {0x5, 0x0, 0x0, 0x0, 0xfe, 0x0, 0x0, 0x2, 0x0};
|
|
||||||
gwbuf_free(pPacket);
|
|
||||||
pPacket = gwbuf_alloc_and_load(sizeof(eof_packet), eof_packet);
|
|
||||||
send = true;
|
send = true;
|
||||||
gwbuf_free(m_query);
|
gwbuf_free(m_query);
|
||||||
m_query = NULL;
|
m_query = NULL;
|
||||||
}
|
}
|
||||||
else if (propagate)
|
|
||||||
{
|
|
||||||
(*m_current)->write(gwbuf_clone(m_query));
|
|
||||||
}
|
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
send = true;
|
(*m_current)->write(gwbuf_clone(m_query));
|
||||||
gwbuf_free(m_query);
|
(*m_current)->set_reply_state(REPLY_STATE_START);
|
||||||
m_query = NULL;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -119,6 +104,7 @@ void CatSession::clientReply(GWBUF* pPacket, DCB* pDcb)
|
|||||||
|
|
||||||
if (send)
|
if (send)
|
||||||
{
|
{
|
||||||
|
// Increment the packet sequence number and send it to the client
|
||||||
GWBUF_DATA(pPacket)[3] = m_packet_num++;
|
GWBUF_DATA(pPacket)[3] = m_packet_num++;
|
||||||
MXS_SESSION_ROUTE_REPLY(pDcb->session, pPacket);
|
MXS_SESSION_ROUTE_REPLY(pDcb->session, pPacket);
|
||||||
}
|
}
|
||||||
@ -130,5 +116,10 @@ void CatSession::clientReply(GWBUF* pPacket, DCB* pDcb)
|
|||||||
|
|
||||||
void CatSession::handleError(GWBUF* pMessage, DCB* pProblem, mxs_error_action_t action, bool* pSuccess)
|
void CatSession::handleError(GWBUF* pMessage, DCB* pProblem, mxs_error_action_t action, bool* pSuccess)
|
||||||
{
|
{
|
||||||
|
/**
|
||||||
|
* The simples thing to do here is to close the connection. Anything else
|
||||||
|
* would still require extra processing on the client side and reconnecting
|
||||||
|
* will cause things to fix themselves.
|
||||||
|
*/
|
||||||
*pSuccess = false;
|
*pSuccess = false;
|
||||||
}
|
}
|
||||||
|
|||||||
@ -77,6 +77,10 @@ private:
|
|||||||
mxs::SRWBackendList::iterator m_current;
|
mxs::SRWBackendList::iterator m_current;
|
||||||
GWBUF* m_query;
|
GWBUF* m_query;
|
||||||
|
|
||||||
// Skip over unused backend servers
|
/**
|
||||||
void skip_unused();
|
* Iterate to next backend
|
||||||
|
*
|
||||||
|
* @return True if m_current points to a valid backend that is in use
|
||||||
|
*/
|
||||||
|
bool next_backend();
|
||||||
};
|
};
|
||||||
Reference in New Issue
Block a user