/*******************************************************************************
* Boap.cc Boap RPC protocol
* T.Barnaby, BEAM Ltd, 8/5/03
*
* History:
* 4/2/05: Updated addressing scheme
*******************************************************************************
*/
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <Boap.h>
// Boap NS
#include <BoapnsD.h>
#include <BoapnsC.h>
#define DEBUG 0
#if DEBUG
#define dprintf(fmt, a...) printf(fmt, ##a);
#else
#define dprintf(fmt, a...)
#endif
const int roundSize = 256;
BoapPacket::BoapPacket(){
opos = 0;
osize = roundSize;
odata = (char*)malloc(osize);
}
BoapPacket::~BoapPacket(){
delete odata;
}
int BoapPacket::resize(int size){
if(size > osize){
osize = roundSize * ((size + roundSize - 1) / roundSize);
odata = (char*)realloc(odata, osize);
}
return 0;
}
BError BoapPacket::setData(void* data, int nbytes){
BError err;
resize(nbytes);
memcpy(odata, data, nbytes);
return err;
}
int BoapPacket::nbytes(){
return ((BoapPacketHead*)odata)->length;
}
char* BoapPacket::data(){
return odata;
}
void BoapPacket::updateLen(){
((BoapPacketHead*)odata)->length = opos;
}
int BoapPacket::pushHead(BoapPacketHead& head){
opos = 0;
resize(opos + sizeof(head));
memcpy(&odata[opos], &head, sizeof(head));
opos += sizeof(head);
updateLen();
return 0;
}
int BoapPacket::push(Int8 v){
resize(opos + sizeof(v));
memcpy(&odata[opos], &v, sizeof(v));
opos += sizeof(v);
updateLen();
return 0;
}
int BoapPacket::push(UInt8 v){
resize(opos + sizeof(v));
memcpy(&odata[opos], &v, sizeof(v));
opos += sizeof(v);
updateLen();
return 0;
}
int BoapPacket::push(Int16 v){
resize(opos + sizeof(v));
memcpy(&odata[opos], &v, sizeof(v));
opos += sizeof(v);
updateLen();
return 0;
}
int BoapPacket::push(UInt16 v){
resize(opos + sizeof(v));
memcpy(&odata[opos], &v, sizeof(v));
opos += sizeof(v);
updateLen();
return 0;
}
int BoapPacket::push(Int32 v){
resize(opos + sizeof(v));
memcpy(&odata[opos], &v, sizeof(v));
opos += sizeof(v);
updateLen();
return 0;
}
int BoapPacket::push(UInt32 v){
resize(opos + sizeof(v));
memcpy(&odata[opos], &v, sizeof(v));
opos += sizeof(v);
updateLen();
return 0;
}
int BoapPacket::push(BString& v){
Int32 l = v.len() + 1;
resize(opos + sizeof(l) + l);
memcpy(&odata[opos], &l, sizeof(l));
opos += sizeof(l);
memcpy(&odata[opos], v.retStr(), l);
opos += l;
updateLen();
return 0;
}
int BoapPacket::push(Double v){
resize(opos + sizeof(v));
memcpy(&odata[opos], &v, sizeof(v));
opos += sizeof(v);
updateLen();
return 0;
}
int BoapPacket::push(BError& v){
Int32 n = v.getErrorNo();
BString s = v.getString();
push(n);
push(s);
return 0;
}
int BoapPacket::push(UInt32 nBytes, const void* data){
}
int BoapPacket::popHead(BoapPacketHead& head){
opos = 0;
memcpy(&head, &odata[opos], sizeof(head));
opos += sizeof(head);
return 0;
}
int BoapPacket::pop(Int8& v){
memcpy(&v, &odata[opos], sizeof(v));
opos += sizeof(v);
return 0;
}
int BoapPacket::pop(UInt8& v){
memcpy(&v, &odata[opos], sizeof(v));
opos += sizeof(v);
return 0;
}
int BoapPacket::pop(Int16& v){
memcpy(&v, &odata[opos], sizeof(v));
opos += sizeof(v);
return 0;
}
int BoapPacket::pop(UInt16& v){
memcpy(&v, &odata[opos], sizeof(v));
opos += sizeof(v);
return 0;
}
int BoapPacket::pop(Int32& v){
memcpy(&v, &odata[opos], sizeof(v));
opos += sizeof(v);
return 0;
}
int BoapPacket::pop(UInt32& v){
memcpy(&v, &odata[opos], sizeof(v));
opos += sizeof(v);
return 0;
}
int BoapPacket::pop(BString& v){
Int32 l;
memcpy(&l, &odata[opos], sizeof(l));
opos += sizeof(l);
v = (char*)&odata[opos];
opos += l;
return 0;
}
int BoapPacket::pop(Double& v){
memcpy(&v, &odata[opos], sizeof(v));
opos += sizeof(v);
return 0;
}
int BoapPacket::pop(BError& v){
Int32 n;
BString s;
pop(n);
pop(s);
v.set(n, s);
return 0;
}
int BoapPacket::pop(UInt32 nBytes, void* data){
return 0;
}
/*******************************************************************************
* Boap Client communications object
*******************************************************************************
*/
/*******************************************************************************
* Base for all Client Classes
*******************************************************************************
*/
BoapClientObject::BoapClientObject(BString name): BSocket(STREAM), oname(name){
oconnected = 0;
oservice = 0;
}
BError BoapClientObject::connectService(BString name){
BError err;
Boapns::Boapns boapns("boapns");
Boapns::BoapEntry entry;
BString host = "localhost";
BString boapnsName;
BSocketAddressINET add;
if(oconnected)
return err;
oname = name;
// Parse name
if(name.subString(0, 2) == "//"){
name.pullSeparators("/");
host = name.pullToken("/");
}
if(name == "boapns"){
if(! (err = add.set(host, "boapns", "tcp"))){
if(! (err = connect(add))){
oservice = 0;
oconnected = 1;
}
}
}
else {
// Lookup name
boapnsName = BString("//") + host + "/boapns";
if(! (err = boapns.connectService(boapnsName))){
if(! (err = boapns.getEntry(name, entry))){
if(! (err = add.set(entry.ipAddress, entry.port))){
if(! (err = connect(add))){
oservice = entry.service;
oconnected = 1;
}
}
}
}
}
return err;
}
BError BoapClientObject::performSend(BoapPacket& tx){
BError err;
BSize nb = tx.nbytes();
char* data = tx.data();
BSize n = 0;
BSize nd = 0;
// Connect if required
if(err = connectService(oname))
return err;
// Send packet
while(n < nb){
if(err = send(&data[n], nb - n, nd, MSG_NOSIGNAL))
return err;
n = n + nd;
}
return err;
}
BError BoapClientObject::performRecv(BoapPacket& rx){
BError err;
BSize nb = 0;
char* data = 0;
BSize n = 0;
BSize nd = 0;
BoapPacketHead head;
// Connect if required
if(err = connectService(oname))
return err;
// Recieve Packet
// Read Head first
rx.resize(sizeof(BoapPacketHead));
nb = sizeof(BoapPacketHead);
data = rx.data();
n = 0;
while(n < nb){
if(err = recv(&data[n], nb - n, nd))
return err;
n = n + nd;
}
// Read rest of packet
rx.popHead(head);
rx.resize(head.length);
nb = head.length - sizeof(BoapPacketHead);
data = &rx.data()[sizeof(BoapPacketHead)];
n = 0;
while(n < nb){
if(err = recv(&data[n], nb - n, nd))
return err;
n = n + nd;
}
return err;
}
BError BoapClientObject::performCall(BoapPacket& tx, BoapPacket& rx){
BError err;
if(! (err = performSend(tx))){
err = performRecv(rx);
}
return err;
}
BoapSignalObject::BoapSignalObject(): BSocket(DGRAM){
setBroadCast(1);
}
BError BoapSignalObject::performSend(BoapPacket& tx){
BError err;
BSize nb = tx.nbytes();
char* data = tx.data();
BSize n = 0;
BSize nd = 0;
BSocketAddressINET nadd;
nadd.set(INADDR_BROADCAST, 14000);
// Send packet
while(n < nb){
if(err = sendTo(nadd, &data[n], nb - n, nd, MSG_NOSIGNAL))
return err;
n = n + nd;
}
return err;
}
/*******************************************************************************
* Boap Server communications object
*******************************************************************************
*/
BoapServer::BoapServer() : onet(BSocket::STREAM), onetEvent(BSocket::DGRAM) {
oboapNs = 0;
onet.setReuseAddress(1);
}
BError BoapServer::init(int boapNs){
BError err;
Boapns::Boapns boapns("boapns");
BIter i;
BoapService s;
BList<uint32_t> l;
uint32_t ipAddress;
BSocketAddressINET nadd;
oboapNs = boapNs;
ohostName = BSocketAddressINET::getHostName();
l = BSocketAddressINET::getIpAddresses();
if(l.number())
ipAddress = l[0];
else
ipAddress = 0;
if(boapNs)
nadd.set("", "boapns", "tcp");
else
nadd.set("", 0);
// Create service socket
if(err = onet.bind(nadd))
return err;
// Create event socket
if(err = onet.getAddress(nadd))
return err;
if(err = onetEvent.bind(nadd))
return err;
// BOAP event broadcast address
onetEventAddress.set("", "boapns", "udp");
onetEventAddress.set(INADDR_BROADCAST, onetEventAddress.port());
onetEvent.setBroadCast(1);
if(!boapNs){
for(oservices.start(i), s = 0; !oservices.isEnd(i); oservices.next(i), s++){
if(err = boapns.addEntry(Boapns::BoapEntry(oservices[i].oobject->name(), ipAddress, nadd.port(), s)))
break;
}
}
return err;
}
BError BoapServer::run(){
BError err;
BSocketAddressINET nadd;
int fd;
opoll.append(onet.getFd());
opoll.append(onetEvent.getFd());
onet.listen();
while(1){
opoll.doPoll(fd);
if(fd == onet.getFd()){
onet.accept(fd, nadd);
dprintf("Accept Connection: Fd: %d Add: %s\n", fd, nadd.getString().retStr());
opoll.append(fd);
}
else if(fd == onetEvent.getFd()){
processEvent(fd);
}
else if(fd >= 0){
if(process(fd)){
opoll.delFd(fd);
close(fd);
}
}
}
return err;
}
BError BoapServer::addObject(BoapServiceObject* object){
BError err;
oservices.append(BoapServiceEntry(oservices.number(), object));
return err;
}
BError BoapServer::process(int fd){
BError err;
int nb = 0;
char* data = 0;
int n = 0;
int nd = 0;
BoapPacketHead head;
BIter i;
// Recieve Packet
// Read Head first
orx.resize(sizeof(BoapPacketHead));
nb = sizeof(BoapPacketHead);
data = orx.data();
n = 0;
while(n < nb){
nd = recv(fd, &data[n], nb - n, 0);
if(nd <= 0)
return BError(1, strerror(errno));
n = n + nd;
}
// Read rest of packet
orx.popHead(head);
orx.resize(head.length);
nb = head.length - sizeof(BoapPacketHead);
data = &orx.data()[sizeof(BoapPacketHead)];
n = 0;
while(n < nb){
nd = recv(fd, &data[n], nb - n, 0);
if(nd <= 0)
return BError(1, strerror(errno));
n = n + nd;
}
dprintf("Command received: Len(%d) Type(%d) Service(%d) Cmd(%d)\n", head.length, head.type, head.service, head.cmd);
for(oservices.start(i); !oservices.isEnd(i); oservices.next(i)){
if(oservices[i].oservice == head.service){
oservices[i].oobject->process(orx, otx);
break;
}
}
// Send packet
nb = otx.nbytes();
data = otx.data();
n = 0;
while(n < nb){
nd = send(fd, &data[n], nb - n, MSG_NOSIGNAL);
if(nd <= 0)
return BError(1, strerror(errno));
n = n + nd;
}
return err;
}
BError BoapServer::processEvent(BoapPacket& rx){
BError err;
BIter i;
int s;
dprintf("BoapServer::processEvent\n");
for(oservices.start(i), s = 0; !oservices.isEnd(i); oservices.next(i), s++){
oservices[i].oobject->processEvent(rx);
}
return err;
}
BError BoapServer::processEvent(int fd){
BError err;
int nb = 0;
BoapPacketHead head;
BoapPacket rx;
// Recieve Packet
rx.resize(1500);
if((nb = recv(fd, rx.data(), 1500, 0)) < 0)
return BError(errno, strerror(errno));
orx.popHead(head);
dprintf("Event received: Len(%d) Service(%d) Cmd(%d)\n", head.length, head.service, head.cmd);
return processEvent(rx);
}
BError BoapServer::sendEvent(BoapPacket& tx){
BError err;
int nb = 0;
char* data = 0;
int n = 0;
uint32_t nsent;
// Send packet
nb = tx.nbytes();
data = tx.data();
n = 0;
while(n < nb){
dprintf("BoapServer::sendEvent: Address: %s\n", onetEventAddress.getString().retStr());
err = onetEvent.sendTo(onetEventAddress, &data[n], nb - n, nsent);
dprintf("BoapServer::sendEvent: result: %d:%s\n", err.getErrorNo(), err.getString().retStr());
if(err)
return err;
n = n + nsent;
}
return err;
}
BSocket& BoapServer::getSocket(){
return onet;
}
BSocket& BoapServer::getEventSocket(){
return onetEvent;
}
BString BoapServer::getHostName(){
return ohostName;
}
/*******************************************************************************
* Base for all Server Objects
*******************************************************************************
*/
BoapFuncEntry::BoapFuncEntry(int cmd, BoapFunc func) :
ocmd(cmd), ofunc(func){
}
BoapServiceObject::BoapServiceObject(BoapServer& server, BString name) : oserver(server), oname(name){
oserver.addObject(this);
}
BoapServiceObject::~BoapServiceObject(){
}
BString BoapServiceObject::name(){
return oname;
}
BError BoapServiceObject::process(BoapPacket& rx, BoapPacket& tx){
BError err;
BoapPacketHead head;
BIter i;
rx.popHead(head);
for(ofuncList.start(i); !ofuncList.isEnd(i); ofuncList.next(i)){
if(ofuncList[i].ocmd == head.cmd){
err = (this->*ofuncList[i].ofunc)(rx, tx);
break;
}
}
return err;
}
BError BoapServiceObject::sendEvent(BString name, Int32 arg){
BError err;
BoapPacketHead txhead;
BoapPacket tx;
BString fullName = BString("//") + oserver.getHostName() + "/" + oname;
txhead.type = BoapTypeSignal;
txhead.service = 0;
txhead.cmd = 0;
tx.pushHead(txhead);
tx.push(fullName);
tx.push(name);
tx.push(arg);
if(err = sendEvent(tx))
return err;
return err;
}
BError BoapServiceObject::processEvent(BString objectName, BString signalName, Int32 arg){
BError err;
dprintf("BoapServiceObject::processEvent(BString objectName, BString signalName, Int32 arg)\n");
return err;
}
BError BoapServiceObject::sendEvent(BoapPacket& tx){
return oserver.sendEvent(tx);
}
BError BoapServiceObject::processEvent(BoapPacket& rx){
BError err;
BoapPacketHead rxhead;
BError ret;
BString objectName;
BString signalName;
Int32 arg;
dprintf("BoapServiceObject::processEvent\n");
rx.popHead(rxhead);
rx.pop(objectName);
rx.pop(signalName);
rx.pop(arg);
ret = processEvent(objectName, signalName, arg);
return err;
}
