Skip to content

Commit

Permalink
StreamConnectionReader - move implementation to cpp file
Browse files Browse the repository at this point in the history
  • Loading branch information
@damn1 @drdanz
damn1 authored and drdanz committed May 4, 2018
1 parent 0116d40 commit 3dac873
Show file tree
Hide file tree
Showing 2 changed files with 345 additions and 330 deletions.
342 changes: 32 additions & 310 deletions src/libYARP_OS/include/yarp/os/impl/StreamConnectionReader.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,326 +38,48 @@ namespace yarp {
class YARP_OS_impl_API yarp::os::impl::StreamConnectionReader : public ConnectionReader
{
public:
StreamConnectionReader() :
ConnectionReader(),
writer(nullptr),
in(nullptr),
str(nullptr),
protocol(nullptr),
messageLen(0),
textMode(false),
bareMode(false),
valid(false),
err(false),
shouldDrop(false),
writePending(false),
ref(nullptr),
convertedTextMode(false),
pushedIntFlag(false),
pushedInt(-1),
parentConnectionReader(nullptr)
{
}

StreamConnectionReader();
virtual ~StreamConnectionReader();

void reset(yarp::os::InputStream& in, TwoWayStream *str, const Route& route,
size_t len, bool textMode, bool bareMode = false)
{
this->in = ∈
this->str = str;
this->route = route;
this->messageLen = len;
this->textMode = textMode;
this->bareMode = bareMode;
this->valid = true;
ref = nullptr;
err = false;
convertedTextMode = false;
pushedIntFlag = false;
}

virtual bool setSize(size_t len) override
{
reset(*in, str, route, len, textMode, bareMode);
return true;
}

void setProtocol(Protocol *protocol)
{
this->protocol = protocol;
}

virtual bool expectBlock(const yarp::os::Bytes& b)
{
if (!isGood()) {
return false;
}
yAssert(in!=nullptr);
size_t len = b.length();
if (len==0) {
return true;
}
//if (len<0) len = messageLen;
if (len>0) {
YARP_SSIZE_T rlen = in->readFull(b);
if (rlen>=0) {
messageLen -= len;
return true;
}
}
err = true;
return false;
}

virtual bool pushInt(int x) override
{
if (pushedIntFlag) {
return false;
}
pushedIntFlag = true;
pushedInt = x;
return true;
}

virtual int expectInt() override
{
if (pushedIntFlag) {
pushedIntFlag = false;
return pushedInt;
}
if (!isGood()) {
return 0;
}
NetInt32 x = 0;
yarp::os::Bytes b((char*)(&x), sizeof(x));
yAssert(in!=nullptr);
YARP_SSIZE_T r = in->read(b);
if (r<0 || (size_t)r<b.length()) {
err = true;
return 0;
}
messageLen -= b.length();
return x;
}

virtual YARP_INT64 expectInt64() override
{
if (!isGood()) {
return 0;
}
NetInt64 x = 0;
yarp::os::Bytes b((char*)(&x), sizeof(x));
yAssert(in!=nullptr);
YARP_SSIZE_T r = in->read(b);
if (r<0 || (size_t)r<b.length()) {
err = true;
return 0;
}
messageLen -= b.length();
return x;
}

virtual double expectDouble() override
{
if (!isGood()) {
return 0;
}
NetFloat64 x = 0;
yarp::os::Bytes b((char*)(&x), sizeof(x));
yAssert(in!=nullptr);
YARP_SSIZE_T r = in->read(b);
if (r<0 || (size_t)r<b.length()) {
err = true;
return 0;
}
messageLen -= b.length();
return x;
}

virtual ConstString expectString(int len)
{
if (!isGood()) {
return "";
}
char *buf = new char[len];
yarp::os::Bytes b(buf, len);
yAssert(in!=nullptr);
YARP_SSIZE_T r = in->read(b);
if (r<0 || (size_t)r<b.length()) {
err = true;
delete[] buf;
return "";
}
messageLen -= b.length();
ConstString s = buf;
delete[] buf;
return s;
}

virtual ConstString expectLine()
{
if (!isGood()) {
return "";
}
yAssert(in!=nullptr);
bool success = false;
ConstString result = in->readLine('\n', &success);
if (!success) {
err = true;
return "";
}
messageLen -= result.length()+1;
return result;
}

virtual bool isTextMode() override
{
return textMode;
}

virtual bool isBareMode() override
{
return bareMode;
}

size_t len, bool textMode, bool bareMode = false);
void setProtocol(Protocol *protocol);
void suppressReply();
bool dropRequested();

virtual bool expectBlock(const yarp::os::Bytes& b);
virtual ConstString expectString(int len);
virtual ConstString expectLine();
virtual void flushWriter();
virtual void setReference(yarp::os::Portable *obj);

/**** OVERRIDES ****/
virtual bool setSize(size_t len) override;
virtual size_t getSize() override;
virtual bool pushInt(int x) override;
virtual int expectInt() override;
virtual YARP_INT64 expectInt64() override;
virtual double expectDouble() override;
virtual bool expectBlock(const char *data, size_t len) override;
virtual ::yarp::os::ConstString expectText(int terminatingChar) override;
virtual bool isTextMode() override;
virtual bool isBareMode() override;
virtual bool convertTextMode() override;

virtual size_t getSize() override
{
return messageLen + (pushedIntFlag?sizeof(yarp::os::NetInt32):0);
}

/*
virtual OutputStream *getReplyStream() override
{
if (str==nullptr) {
return nullptr;
}
return &(str->getOutputStream());
}
*/

virtual yarp::os::ConnectionWriter *getWriter() override;

void suppressReply()
{
str = nullptr;
}

virtual void flushWriter();

// virtual TwoWayStream *getStreams() override
// {
// return str;
// }

virtual yarp::os::Contact getRemoteContact() override
{
if (str!=nullptr) {
Contact remote = str->getRemoteAddress();
remote.setName(route.getFromName());
return remote;
}
Contact remote = yarp::os::Contact(route.getFromName(), route.getCarrierName());
return remote;
}

virtual yarp::os::Contact getLocalContact() override
{
if (str!=nullptr) {
Contact local = str->getLocalAddress();
local.setName(route.getToName());
return local;
}
return yarp::os::Contact();
}



virtual bool expectBlock(const char *data, size_t len) override
{
return expectBlock(yarp::os::Bytes((char*)data, len));
}

virtual ::yarp::os::ConstString expectText(int terminatingChar) override
{
if (!isGood()) {
return "";
}
yAssert(in!=nullptr);
bool lsuccess = false;
ConstString result = in->readLine(terminatingChar, &lsuccess);
if (lsuccess) {
messageLen -= result.length()+1;
}
return ::yarp::os::ConstString(result.c_str());
}

virtual bool isValid() override
{
return valid;
}

virtual bool isError() override
{
if (err) {
return true;
}
return !isActive();
}

virtual bool isActive() override
{
if (shouldDrop) {
return false;
}
if (!isValid()) {
return false;
}
if (in!=nullptr) {
if (in->isOk()) {
return true;
}
}
return false;
}

virtual yarp::os::Portable *getReference() override
{
return ref;
}

virtual void setReference(yarp::os::Portable *obj)
{
ref = obj;
}

virtual yarp::os::Contact getRemoteContact() override;
virtual yarp::os::Contact getLocalContact() override;
virtual bool isValid() override;
virtual bool isError() override;
virtual bool isActive() override;
virtual yarp::os::Portable *getReference() override;
virtual yarp::os::Bytes readEnvelope() override;

virtual void requestDrop() override
{
shouldDrop = true;
}

bool dropRequested()
{
return shouldDrop;
}

virtual void requestDrop() override;
virtual yarp::os::Searchable& getConnectionModifiers() override;

virtual void setParentConnectionReader(ConnectionReader *parentConnectionReader) override
{
this->parentConnectionReader = parentConnectionReader;
}
virtual void setParentConnectionReader(ConnectionReader *parentConnectionReader) override;

private:

bool isGood()
{
return isActive()&&isValid()&&!isError();
}
bool isGood();

BufferedConnectionWriter *writer;
StringInputStream altStream;
Expand Down
Loading

0 comments on commit 3dac873

Please sign in to comment.