epics/pvAccess
0
0
Fork 0
Code Issues Pull Requests Projects Wiki Activity

blockingServerTCPTransport.cpp:

Browse Source 
- using enum to specify command.

BlockingTCPAcceptor.cpp:
- added 'destroy()' to dtor
- added parentheses to expressions

blockingTCPCinnector.cpp:
- fixed log message

blockingTCPTransport.cpp:
- added _socketAddress allocation
- fixed waiting for timeout in 'waitUntilVerified'
- fixed how many bytes are copied from the buffer

responseHandlers.cpp:
- added 'ConnectionValidationHandler' implementation
- added 'ConnectionValidationHandler' to 'ServerResponseHandler'

responseHandlers.h:
- added 'ConnectionValidationHandler' declaration

inetAddressUtil.cpp:
- fixed all issues with sockaddr_in byte-order
- removed function 'processAddressForList', using EPICSv3 'aToIPAddr' instead

inetAddressUtilsTest.cpp:
- fixed the tests in accordance with the correct function implementation

testBlockingUDPClnt.cpp:
- deleting transport at the end of the test

testBlockingTCPSrv.cpp, testBlockingTCPClnt.cpp:
- added tests (work in progress).

Makefile:
- added blockingTCP tests
This commit is contained in:
miha_vitorovic
2011-01-06 14:58:32 +01:00
parent 6c3b00dc86
commit 8339c338de
12 changed files with 282 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pvAccessApp/remote/blockingServerTCPTransport.cpp
View File

@@ -111,7 +111,7 @@ namespace epics {
//
// send verification message
//
control->startMessage(1, 2*sizeof(int32));
control->startMessage(CMD_CONNECTION_VALIDATION, 2*sizeof(int32));
// receive buffer size
buffer->putInt(getReceiveBufferSize());

6
pvAccessApp/remote/blockingTCPAcceptor.cpp
View File

@@ -40,6 +40,8 @@ namespace epics {
}
BlockingTCPAcceptor::~BlockingTCPAcceptor() {
destroy();
if(_bindAddress!=NULL) delete _bindAddress;
_connectedClientsMutex->lock();
@@ -190,7 +192,7 @@ namespace epics {
}
else if(retval>0) {
// some event on a socket
if(sockets[0].revents&POLLIN!=0) {
if((sockets[0].revents&POLLIN)!=0) {
// connection waiting
osiSockAddr address;
@@ -267,7 +269,7 @@ namespace epics {
}// accept succeeded
} // connection waiting
if(sockets[0].revents&(POLLERR|POLLHUP|POLLNVAL)!=0) {
if((sockets[0].revents&(POLLERR|POLLHUP|POLLNVAL))!=0) {
errlogSevPrintf(errlogMajor,
"error on a socket: POLLERR|POLLHUP|POLLNVAL");
socketOpen = false;

2
pvAccessApp/remote/blockingTCPConnector.cpp
View File

@@ -139,7 +139,7 @@ namespace epics {
transport->close(true);
errlogSevPrintf(
errlogInfo,
"Connection to CA client %s failed to be validated, closing it.",
"Connection to CA server %s failed to be validated, closing it.",
ipAddrStr);
ostringstream temp;
temp<<"Failed to verify TCP connection to '"<<ipAddrStr

9
pvAccessApp/remote/blockingTCPTransport.cpp
View File

@@ -65,7 +65,8 @@ namespace epics {
BlockingTCPTransport::BlockingTCPTransport(Context* context,
SOCKET channel, ResponseHandler* responseHandler,
int receiveBufferSize, int16 priority) :
_closed(false), _channel(channel), _remoteTransportRevision(0),
_closed(false), _channel(channel), _socketAddress(new osiSockAddr),
_remoteTransportRevision(0),
_remoteTransportReceiveBufferSize(MAX_TCP_RECV),
_remoteTransportSocketReceiveBufferSize(MAX_TCP_RECV),
_priority(priority), _responseHandler(responseHandler),
@@ -99,7 +100,6 @@ namespace epics {
// get send buffer size
socklen_t intLen = sizeof(int);
int retval = getsockopt(_channel, SOL_SOCKET, SO_SNDBUF,
@@ -129,6 +129,7 @@ namespace epics {
BlockingTCPTransport::~BlockingTCPTransport() {
close(true);
delete _socketAddress;
delete _sendQueue;
delete _socketBuffer;
delete _sendBuffer;
@@ -236,7 +237,7 @@ namespace epics {
internalVerified = _verified;
_verifiedMutex->unlock();
while(!internalVerified&&internalTimeout<=0) {
while(!internalVerified&&internalTimeout>0) {
epicsThreadSleep(min(0.1, internalTimeout));
internalTimeout -= 0.1;
@@ -459,7 +460,7 @@ namespace epics {
_socketBuffer->getRemaining());
ssize_t bytesRead = recv(_channel, readBuffer,
maxToRead, 0);
_socketBuffer->put(readBuffer, 0, maxToRead);
_socketBuffer->put(readBuffer, 0, bytesRead);
if(bytesRead<0) {
// error (disconnect, end-of-stream) detected

20
pvAccessApp/server/responseHandlers.cpp
View File

@@ -64,7 +64,7 @@ namespace epics {
_handlerTable = new ResponseHandler*[HANDLER_TABLE_LENGTH];
// TODO add real handlers, as they are developed
_handlerTable[0] = badResponse;
_handlerTable[1] = badResponse;
_handlerTable[1] = new ConnectionValidationHandler(_context);
_handlerTable[2] = badResponse;
_handlerTable[3] = badResponse;
_handlerTable[4] = badResponse;
@@ -95,6 +95,7 @@ namespace epics {
ServerResponseHandler::~ServerResponseHandler() {
delete _handlerTable[0];
delete _handlerTable[1];
delete[] _handlerTable;
}
@@ -119,5 +120,22 @@ namespace epics {
version, command, payloadSize, payloadBuffer);
}
void ConnectionValidationHandler::handleResponse(
osiSockAddr* responseFrom, Transport* transport, int8 version,
int8 command, int payloadSize,
epics::pvData::ByteBuffer* payloadBuffer) {
AbstractServerResponseHandler::handleResponse(responseFrom,
transport, version, command, payloadSize, payloadBuffer);
transport->ensureData(2*sizeof(int32)+sizeof(int16));
transport->setRemoteTransportReceiveBufferSize(
payloadBuffer->getInt());
transport->setRemoteTransportSocketReceiveBufferSize(
payloadBuffer->getInt());
transport->setRemoteMinorRevision(version);
// TODO support priority !!!
//transport.setPriority(payloadBuffer.getShort());
}
}
}

18
pvAccessApp/server/responseHandlers.h
View File

@@ -84,10 +84,26 @@ namespace epics {
};
/**
* Connection validation message handler.
* @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
* @version $Id: ConnectionValidationHandler.java,v 1.1 2010/05/03 14:45:39 mrkraimer Exp $
*/
class ConnectionValidationHandler : public AbstractServerResponseHandler {
public:
/**
* @param context
*/
ConnectionValidationHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Connection validation") {
}
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
};
}
}
#endif /* RESPONSEHANDLERS_H_ */

39
pvAccessApp/utils/inetAddressUtil.cpp
View File

@@ -21,6 +21,7 @@
#include <cstring>
#include <cstdlib>
#include <sstream>
#include <sys/socket.h>
using namespace std;
using namespace epics::pvData;
@@ -124,14 +125,14 @@ namespace epics {
osiSockAddr* intToIPv4Address(int32 addr) {
osiSockAddr* ret = new osiSockAddr;
ret->ia.sin_family = AF_INET;
ret->ia.sin_addr.s_addr = (in_addr_t)addr;
ret->ia.sin_addr.s_addr = htonl(addr);
ret->ia.sin_port = 0;
return ret;
}
int32 ipv4AddressToInt(const osiSockAddr& addr) {
return (int32)(addr.ia.sin_addr.s_addr);
return (int32)ntohl(addr.ia.sin_addr.s_addr);
}
int32 parseInetAddress(const String addr) {
@@ -168,24 +169,6 @@ namespace epics {
return retAddr;
}
osiSockAddr* processAddressForList(String address, int defaultPort) {
// check port
int port = defaultPort;
size_t pos = address.find(':');
if(pos!=String::npos) {
port = atoi(address.substr(pos+1).c_str());
address = address.substr(0, pos);
}
// add parsed address
osiSockAddr* addr = new osiSockAddr;
addr->ia.sin_family = AF_INET;
addr->ia.sin_port = port;
addr->ia.sin_addr.s_addr = parseInetAddress(address);
return addr;
}
InetAddrVector* getSocketAddressList(String list, int defaultPort,
const InetAddrVector* appendList) {
InetAddrVector* iav = new InetAddrVector();
@@ -195,13 +178,17 @@ namespace epics {
size_t subEnd;
while((subEnd = list.find(' ', subStart))!=String::npos) {
String address = list.substr(subStart, (subEnd-subStart));
iav->push_back(processAddressForList(address, defaultPort));
osiSockAddr* addr = new osiSockAddr;
aToIPAddr(address.c_str(), defaultPort, &addr->ia);
iav->push_back(addr);
subStart = list.find_first_not_of(" \t\r\n\v", subEnd);
}
if(subStart!=String::npos&&list.length()>0) iav->push_back(
processAddressForList(list.substr(subStart), defaultPort));
if(subStart!=String::npos&&list.length()>0) {
osiSockAddr* addr = new osiSockAddr;
aToIPAddr(list.substr(subStart).c_str(), defaultPort, &addr->ia);
iav->push_back(addr);
}
if(appendList!=NULL) {
for(size_t i = 0; i<appendList->size(); i++)
@@ -221,8 +208,8 @@ namespace epics {
saddr<<((int)(ipa>>8)&0xFF)<<'.';
saddr<<((int)ipa&0xFF);
if(addr->ia.sin_port>0) saddr<<":"<<ntohs(addr->ia.sin_port);
if(displayHex) saddr<<" ("<<hex<<((uint32_t)(
addr->ia.sin_addr.s_addr))<<")";
if(displayHex) saddr<<" ("<<hex
<<ntohl(addr->ia.sin_addr.s_addr)<<")";
return saddr.str();
}

8
testApp/remote/Makefile
View File

@@ -22,6 +22,14 @@ PROD_HOST += testBeaconHandler
testBeaconHandler_SRCS += testBeaconHandler.cpp
testBeaconHandler_LIBS += pvData pvAccess Com
PROD_HOST += testBlockingTCPSrv
testBlockingTCPSrv_SRCS += testBlockingTCPSrv.cpp
testBlockingTCPSrv_LIBS += pvData pvAccess Com
PROD_HOST += testBlockingTCPClnt
testBlockingTCPClnt_SRCS += testBlockingTCPClnt
testBlockingTCPClnt_LIBS += pvData pvAccess Com
include $(TOP)/configure/RULES
#----------------------------------------
# ADD RULES AFTER THIS LINE

134
testApp/remote/testBlockingTCPClnt.cpp Normal file
View File

@@ -0,0 +1,134 @@
/*
* testBlockingTCPClnt.cpp
*
* Created on: Jan 6, 2011
* Author: Miha Vitorovic
*/
#include "remote.h"
#include "blockingTCP.h"
#include "logger.h"
#include "inetAddressUtil.h"
#include "caConstants.h"
#include <timer.h>
#include <osiSock.h>
#include <errlog.h>
#include <iostream>
#include <cstdio>
using namespace epics::pvAccess;
using namespace epics::pvData;
using std::cout;
using std::endl;
using std::sscanf;
class ContextImpl : public Context {
public:
ContextImpl() :
_tr(new TransportRegistry()),
_timer(new Timer("client thread", lowPriority)) {}
virtual ~ContextImpl() {
delete _tr;
delete _timer;
}
virtual Timer* getTimer() { return _timer; }
virtual TransportRegistry* getTransportRegistry() { return _tr; }
private:
TransportRegistry* _tr;
Timer* _timer;
};
class DummyResponseHandler : public ResponseHandler {
public:
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command, int payloadSize,
ByteBuffer* payloadBuffer) {
}
};
class DummyTransportClient : public TransportClient {
public:
DummyTransportClient() {
}
virtual ~DummyTransportClient() {
}
virtual void transportUnresponsive() {
errlogSevPrintf(errlogInfo, "unresponsive");
}
virtual void transportResponsive(Transport* transport) {
errlogSevPrintf(errlogInfo, "responsive");
}
virtual void transportChanged() {
errlogSevPrintf(errlogInfo, "changed");
}
virtual void transportClosed() {
errlogSevPrintf(errlogInfo, "closed");
}
};
class DummyTransportSender : public TransportSender {
public:
DummyTransportSender() {
for(int i = 0; i<20; i++)
data[i] = (char)(i+1);
count = 0;
}
virtual void send(ByteBuffer* buffer, TransportSendControl* control) {
// send the packet
count++;
control->startMessage(0, count);
buffer->put(data, 0, count);
//control->endMessage();
}
virtual void lock() {
}
virtual void unlock() {
}
private:
char data[20];
int count;
};
void testBlockingTCPSender() {
ContextImpl ctx;
BlockingTCPConnector connector(&ctx, 1024, 1.0);
DummyTransportClient dtc;
DummyTransportSender dts;
DummyResponseHandler drh;
osiSockAddr srvAddr;
srvAddr.ia.sin_family = AF_INET;
//srvAddr.ia.sin_port = htons(CA_SERVER_PORT);
if(aToIPAddr("192.168.71.132", CA_SERVER_PORT, &srvAddr.ia)<0) {
cout<<"error in aToIPAddr(...)"<<endl;
return;
}
Transport* transport = connector.connect(&dtc, &drh, &srvAddr,
CA_MAGIC_AND_VERSION, CA_DEFAULT_PRIORITY);
cout<<"Sending 10 messages..."<<endl;
for(int i = 0; i<10; i++) {
cout<<" Message: "<<i+1<<endl;
transport->enqueueSendRequest(&dts);
sleep(1);
}
delete transport;
}
int main(int argc, char *argv[]) {
createFileLogger("testBlockingTCPClnt.log");
testBlockingTCPSender();
return 0;
}

53
testApp/remote/testBlockingTCPSrv.cpp Normal file
View File

@@ -0,0 +1,53 @@
/*
* testBlockingTCPSrv.cpp
*
* Created on: Jan 6, 2011
* Author: Miha Vitorovic
*/
#include "blockingTCP.h"
#include "remote.h"
#include "logger.h"
#include <iostream>
using namespace epics::pvData;
using namespace epics::pvAccess;
using std::cin;
using std::cout;
class ContextImpl : public Context {
public:
ContextImpl() :
_tr(new TransportRegistry()),
_timer(new Timer("server thread", lowPriority)) {}
virtual ~ContextImpl() {
delete _tr;
delete _timer;
}
virtual Timer* getTimer() { return _timer; }
virtual TransportRegistry* getTransportRegistry() { return _tr; }
private:
TransportRegistry* _tr;
Timer* _timer;
};
void testServerConnections() {
ContextImpl ctx;
BlockingTCPAcceptor* srv = new BlockingTCPAcceptor(&ctx, CA_SERVER_PORT,
1024);
cout<<"Press any key to stop the server...";
char c = cin.peek();
delete srv;
}
int main(int argc, char *argv[]) {
createFileLogger("testBlockingTCPSrv.log");
testServerConnections();
}

2
testApp/remote/testBlockingUDPClnt.cpp
View File

@@ -89,6 +89,8 @@ void testBlockingUDPSender() {
sleep(1);
}
delete transport;
}
int main(int argc, char *argv[]) {

42
testApp/utils/inetAddressUtilsTest.cpp
View File

@@ -32,20 +32,23 @@ int main(int argc, char *argv[]) {
osiSockAddr* addr;
addr = vec->at(0);
assert(addr->ia.sin_family==AF_INET);
assert(addr->ia.sin_port==555);
assert(addr->ia.sin_addr.s_addr==(uint32_t)0x7F000001);
assert(addr->ia.sin_port==htons(555));
assert(addr->ia.sin_addr.s_addr==htonl(0x7F000001));
assert(inetAddressToString(addr)=="127.0.0.1:555");
cout<<'\t'<<inetAddressToString(addr, true)<<endl;
addr = vec->at(1);
assert(addr->ia.sin_family==AF_INET);
assert(addr->ia.sin_port==1234);
assert(addr->ia.sin_addr.s_addr==(uint32_t)0x0A0A0C0B);
assert(addr->ia.sin_port==htons(1234));
assert(addr->ia.sin_addr.s_addr==htonl(0x0A0A0C0B));
assert(inetAddressToString(addr)=="10.10.12.11:1234");
cout<<'\t'<<inetAddressToString(addr, true)<<endl;
addr = vec->at(2);
assert(addr->ia.sin_family==AF_INET);
assert(addr->ia.sin_port==555);
assert(addr->ia.sin_addr.s_addr==(uint32_t)0xC0A80304);
assert(addr->ia.sin_port==htons(555));
assert(addr->ia.sin_addr.s_addr==htonl(0xC0A80304));
assert(inetAddressToString(addr)=="192.168.3.4:555");
cout<<'\t'<<inetAddressToString(addr, true)<<endl;
cout<<"\nPASSED!\n";
@@ -57,26 +60,30 @@ int main(int argc, char *argv[]) {
addr = vec1->at(0);
assert(addr->ia.sin_family==AF_INET);
assert(addr->ia.sin_port==6789);
assert(addr->ia.sin_addr.s_addr==(uint32_t)0xAC1037A0);
cout<<'\t'<<inetAddressToString(addr, true)<<endl;
assert(addr->ia.sin_port==htons(6789));
assert(addr->ia.sin_addr.s_addr==htonl(0xAC1037A0));
assert(inetAddressToString(addr)=="172.16.55.160:6789");
cout<<'\t'<<inetAddressToString(addr, true)<<endl;
addr = vec1->at(1);
assert(addr->ia.sin_family==AF_INET);
assert(addr->ia.sin_port==555);
assert(addr->ia.sin_addr.s_addr==(uint32_t)0x7F000001);
assert(addr->ia.sin_port==htons(555));
assert(addr->ia.sin_addr.s_addr==htonl(0x7F000001));
assert(inetAddressToString(addr)=="127.0.0.1:555");
cout<<'\t'<<inetAddressToString(addr, true)<<endl;
addr = vec1->at(2);
assert(addr->ia.sin_family==AF_INET);
assert(addr->ia.sin_port==1234);
assert(addr->ia.sin_addr.s_addr==(uint32_t)0x0A0A0C0B);
assert(addr->ia.sin_port==htons(1234));
assert(addr->ia.sin_addr.s_addr==htonl(0x0A0A0C0B));
assert(inetAddressToString(addr)=="10.10.12.11:1234");
cout<<'\t'<<inetAddressToString(addr, true)<<endl;
addr = vec1->at(3);
assert(addr->ia.sin_family==AF_INET);
assert(addr->ia.sin_port==555);
assert(addr->ia.sin_addr.s_addr==(uint32_t)0xC0A80304);
assert(addr->ia.sin_port==htons(555));
assert(addr->ia.sin_addr.s_addr==htonl(0xC0A80304));
assert(inetAddressToString(addr)=="192.168.3.4:555");
cout<<'\t'<<inetAddressToString(addr, true)<<endl;
cout<<"\nPASSED!\n";
@@ -93,11 +100,13 @@ int main(int argc, char *argv[]) {
cout<<"Testing \"intToIPv4Address\""<<endl;
addr = intToIPv4Address(0x7F000001);
assert(addr->ia.sin_family==AF_INET);
assert(inetAddressToString(addr)=="127.0.0.1");
cout<<'\t'<<inetAddressToString(addr, true)<<endl;
delete addr;
addr = intToIPv4Address(0x0A0A0C0B);
assert(addr->ia.sin_family==AF_INET);
assert(inetAddressToString(addr)=="10.10.12.11");
cout<<'\t'<<inetAddressToString(addr, true)<<endl;
cout<<"\nPASSED!\n";
@@ -110,9 +119,8 @@ int main(int argc, char *argv[]) {
(char)0, (char)0, (char)0, (char)0, (char)0xFF, (char)0xFF,
(char)0x0A, (char)0x0A, (char)0x0C, (char)0x0B };
encodeAsIPv6Address(buff, addr);
assert(strncmp(buff->getArray(),src,16)==0);
assert(strncmp(buff->getArray(), src, 16)==0);
cout<<"\nPASSED!\n";
// TODO add test for 'getBroadcastAddresses'