moved part of the code to libretroshare/src/friend_server

libretroshare/src/friend_server/fsbio.cc

@@ -0,0 +1,133 @@
+FsBioInterface::FsBioInterface(int socket)
+    : mCLintConnt(socket)
+{
+    mTotalReadBytes=0;
+    mTotalBufferBytes=0;
+}
+
+int FsBioInterface::tick()
+{
+    std::cerr << "ticking FsNetworkInterface" << std::endl;
+
+    // 2 - read incoming data pending on existing connections
+
+    char inBuffer[1025];
+    memset(inBuffer,0,1025);
+
+    int readbytes = read(mCLintConnt, inBuffer, sizeof(inBuffer));
+
+    if(readbytes == 0)
+    {
+        std::cerr << "Reached END of the stream!" << std::endl;
+        return 0;
+    }
+    if(readbytes < 0)
+    {
+        if(errno != EWOULDBLOCK && errno != EAGAIN)
+            RsErr() << "read() failed. Errno=" << errno ;
+
+        return false;
+    }
+
+    std::cerr << "clintConnt: " << mCLintConnt << ", readbytes: " << readbytes << std::endl;
+
+    //::close(clintConnt);
+
+    // display some debug info
+
+    if(readbytes > 0)
+    {
+        RsDbg() << "Received the following bytes: " << RsUtil::BinToHex( reinterpret_cast<unsigned char*>(inBuffer),readbytes,50) << std::endl;
+        //RsDbg() << "Received the following bytes: " << std::string(inBuffer,readbytes) << std::endl;
+
+        void *ptr = malloc(readbytes);
+
+        if(!ptr)
+            throw std::runtime_error("Cannot allocate memory! Go buy some RAM!");
+
+        memcpy(ptr,inBuffer,readbytes);
+
+        in_buffer.push_back(std::make_pair(ptr,readbytes));
+        mTotalBufferBytes += readbytes;
+        mTotalReadBytes += readbytes;
+
+        std::cerr << "Socket: " << mCLintConnt << ". Total read: " << mTotalReadBytes << ". Buffer size: " << mTotalBufferBytes << std::endl ;
+    }
+
+    return true;
+}
+
+int FsBioInterface::readdata(void *data, int len)
+{
+    // read incoming bytes in the buffer
+
+    int total_len = 0;
+
+    while(total_len < len)
+    {
+        if(in_buffer.empty())
+        {
+            mTotalBufferBytes -= total_len;
+            return total_len;
+        }
+
+        // If the remaining buffer is too large, chop of the beginning of it.
+
+        if(total_len + in_buffer.front().second > len)
+        {
+            memcpy(&(static_cast<unsigned char *>(data)[total_len]),in_buffer.front().first,len - total_len);
+
+            void *ptr = malloc(in_buffer.front().second - (len - total_len));
+            memcpy(ptr,&(static_cast<unsigned char*>(in_buffer.front().first)[len - total_len]),in_buffer.front().second - (len - total_len));
+
+            free(in_buffer.front().first);
+            in_buffer.front().first = ptr;
+            in_buffer.front().second -= len-total_len;
+
+            mTotalBufferBytes -= len;
+            return len;
+        }
+        else // copy everything
+        {
+            memcpy(&(static_cast<unsigned char *>(data)[total_len]),in_buffer.front().first,in_buffer.front().second);
+
+            total_len += in_buffer.front().second;
+
+            free(in_buffer.front().first);
+            in_buffer.pop_front();
+        }
+    }
+    mTotalBufferBytes -= len;
+    return len;
+}
+
+int FsBioInterface::senddata(void *data, int len)
+{
+//    int written = write(mCLintConnt, data, len);
+//    return written;
+    return len;
+}
+int FsBioInterface::netstatus()
+{
+    return 1; // dummy response.
+}
+int FsBioInterface::isactive()
+{
+    return mCLintConnt > 0;
+}
+bool FsBioInterface::moretoread(uint32_t /* usec */)
+{
+    return mTotalBufferBytes > 0;
+}
+bool FsBioInterface::cansend(uint32_t)
+{
+    return isactive();
+}
+
+int FsBioInterface::close()
+{
+    RsDbg() << "Stopping network interface" << std::endl;
+    return 1;
+}
+
+

libretroshare/src/friend_server/fsbio.h

@@ -0,0 +1,35 @@
+class FsBioInterface: public BinInterface
+{
+public:
+    FsBioInterface(int socket);
+
+     // Implements BinInterface methods
+
+    int tick() override;
+
+    int senddata(void *data, int len) override;
+    int readdata(void *data, int len) override;
+
+    int netstatus() override;
+    int isactive() override;
+    bool moretoread(uint32_t usec) override;
+    bool cansend(uint32_t usec) override;
+
+    int close() override;
+
+    /**
+     * If hashing data
+     **/
+    RsFileHash gethash() override { return RsFileHash() ; }
+    uint64_t bytecount() override { return mTotalReadBytes; }
+
+    bool bandwidthLimited() override { return false; }
+
+private:
+    int mCLintConnt;
+    uint32_t mTotalReadBytes;
+    uint32_t mTotalBufferBytes;
+
+    std::list<std::pair<void *,int> > in_buffer;
+};
+

libretroshare/src/friend_server/fsclient.cc

@@ -0,0 +1,77 @@
+FsClient::FsClient(const std::string& address)
+    : mServerAddress(address)
+{
+}
+
+bool FsClient::sendItem(RsItem *item)
+{
+    // open a connection
+
+    int CreateSocket = 0,n = 0;
+    char dataReceived[1024];
+    struct sockaddr_in ipOfServer;
+
+    memset(dataReceived, '0' ,sizeof(dataReceived));
+
+    if((CreateSocket = socket(AF_INET, SOCK_STREAM, 0))< 0)
+    {
+        printf("Socket not created \n");
+        return 1;
+    }
+
+    ipOfServer.sin_family = AF_INET;
+    ipOfServer.sin_port = htons(2017);
+    ipOfServer.sin_addr.s_addr = inet_addr("127.0.0.1");
+
+    if(connect(CreateSocket, (struct sockaddr *)&ipOfServer, sizeof(ipOfServer))<0)
+    {
+        printf("Connection failed due to port and ip problems, or server is not available\n");
+        return false;
+    }
+
+    // Serialise the item and send it.
+
+    uint32_t size = RsSerialiser::MAX_SERIAL_SIZE;
+    RsTemporaryMemory data(size);
+
+    if(!data)
+    {
+        RsErr() << "Cannot allocate memory to send item!" << std::endl;
+        return false;
+    }
+
+    FsSerializer *fss = new FsSerializer;
+    RsSerialiser rss;
+    rss.addSerialType(fss);
+
+    FsSerializer().serialise(item,data,&size);
+
+    // TODO: we should write in multiple chunks just in case the socket is not fully ready
+    write(CreateSocket,data,size);
+
+    // Now attempt to read and deserialize anything that comes back from that connexion
+
+    FsBioInterface bio(CreateSocket);
+    pqistreamer pqi(&rss,RsPeerId(),&bio,BIN_FLAGS_READABLE);
+    pqithreadstreamer p(&pqi,&rss,RsPeerId(),&bio,BIN_FLAGS_READABLE);
+    p.start();
+
+    while(true)
+    {
+        RsItem *item = p.GetItem();
+
+        if(!item)
+        {
+            rstime::rs_usleep(1000*200);
+            continue;
+        }
+
+        std::cerr << "Got a response item: " << std::endl;
+        std::cerr << *item << std::endl;
+    }
+
+    return 0;
+
+    // if ok, stream the item through it
+}
+

libretroshare/src/friend_server/fsclient.h

@@ -0,0 +1,13 @@
+// This class runs a client connection to the friend server. It opens a socket at each connection.
+
+class FsClient
+{
+public:
+    FsClient(const std::string& address);
+
+    bool sendItem(RsItem *item);
+
+private:
+    std::string mServerAddress;
+};
+

libretroshare/src/friend_server/fsitem.h

@@ -0,0 +1,118 @@
+#include "serialiser/rsserial.h"
+#include "serialiser/rsserializer.h"
+
+#include "rsitems/rsitem.h"
+#include "rsitems/rsserviceids.h"
+#include "rsitems/itempriorities.h"
+
+const uint8_t RS_PKT_SUBTYPE_FS_CLIENT_PUBLISH             = 0x01 ;
+const uint8_t RS_PKT_SUBTYPE_FS_CLIENT_REMOVE              = 0x02 ;
+const uint8_t RS_PKT_SUBTYPE_FS_SERVER_RESPONSE            = 0x03 ;
+const uint8_t RS_PKT_SUBTYPE_FS_SERVER_ENCRYPTED_RESPONSE  = 0x04 ;
+
+class RsFriendServerItem: public RsItem
+{
+public:
+    RsFriendServerItem(uint8_t item_subtype) : RsItem(RS_PKT_VERSION_SERVICE,RS_SERVICE_TYPE_FRIEND_SERVER,item_subtype)
+	{
+		setPriorityLevel(QOS_PRIORITY_DEFAULT) ;
+	}
+    virtual ~RsFriendServerItem() {}
+    virtual void clear()  override {}
+};
+
+class RsFriendServerClientPublishItem: public RsFriendServerItem
+{
+public:
+    RsFriendServerClientPublishItem() : RsFriendServerItem(RS_PKT_SUBTYPE_FS_CLIENT_PUBLISH) {}
+
+    void serial_process(RsGenericSerializer::SerializeJob j,RsGenericSerializer::SerializeContext& ctx) override
+    {
+        RS_SERIAL_PROCESS(n_requested_friends);
+        RS_SERIAL_PROCESS(long_invite);
+    }
+    virtual void clear()  override
+    {
+        long_invite = std::string();
+        n_requested_friends=0;
+    }
+
+    // specific members for that item
+
+    uint32_t    n_requested_friends;
+    std::string long_invite;
+};
+
+class RsFriendServerClientRemoveItem: public RsFriendServerItem
+{
+public:
+    RsFriendServerClientRemoveItem() : RsFriendServerItem(RS_PKT_SUBTYPE_FS_CLIENT_REMOVE) {}
+
+    void serial_process(RsGenericSerializer::SerializeJob /* j */,RsGenericSerializer::SerializeContext& /* ctx */)
+    {
+    }
+};
+class RsFriendServerEncryptedServerResponseItem: public RsFriendServerItem
+{
+public:
+    RsFriendServerEncryptedServerResponseItem() : RsFriendServerItem(RS_PKT_SUBTYPE_FS_SERVER_ENCRYPTED_RESPONSE) {}
+
+    void serial_process(RsGenericSerializer::SerializeJob j,RsGenericSerializer::SerializeContext& ctx) override
+    {
+        RsTypeSerializer::RawMemoryWrapper prox(bin_data, bin_len);
+        RsTypeSerializer::serial_process(j, ctx, prox, "data");
+    }
+
+    virtual void clear() override
+    {
+        free(bin_data);
+        bin_len = 0;
+        bin_data = nullptr;
+    }
+    //
+
+    void *bin_data;
+    uint32_t bin_len;
+};
+
+class RsFriendServerServerResponseItem: public RsFriendServerItem
+{
+public:
+    RsFriendServerServerResponseItem() : RsFriendServerItem(RS_PKT_SUBTYPE_FS_SERVER_RESPONSE) {}
+
+    void serial_process(RsGenericSerializer::SerializeJob j,RsGenericSerializer::SerializeContext& ctx) override
+    {
+        RS_SERIAL_PROCESS(friend_invites);
+    }
+
+    virtual void clear() override
+    {
+        friend_invites.clear();
+    }
+    // specific members for that item
+
+    std::map<std::string,bool> friend_invites;
+};
+
+struct FsSerializer : RsServiceSerializer
+{
+    FsSerializer(RsSerializationFlags flags = RsSerializationFlags::NONE): RsServiceSerializer(RS_SERVICE_TYPE_FRIEND_SERVER, flags) {}
+
+    virtual RsItem *create_item(uint16_t service_id,uint8_t item_sub_id) const
+    {
+        if(service_id != static_cast<uint16_t>(RsServiceType::FRIEND_SERVER))
+            return nullptr;
+
+        switch(item_sub_id)
+        {
+        case RS_PKT_SUBTYPE_FS_CLIENT_REMOVE:             return new RsFriendServerClientRemoveItem();
+        case RS_PKT_SUBTYPE_FS_CLIENT_PUBLISH:            return new RsFriendServerClientPublishItem();
+        case RS_PKT_SUBTYPE_FS_SERVER_RESPONSE:           return new RsFriendServerServerResponseItem();
+        case RS_PKT_SUBTYPE_FS_SERVER_ENCRYPTED_RESPONSE: return new RsFriendServerEncryptedServerResponseItem();
+        default:
+            RsErr() << "Unknown subitem type " << item_sub_id << " in FsSerialiser" ;
+            return nullptr;
+        }
+
+    }
+};

libretroshare/src/libretroshare.pro

@@ -404,6 +404,10 @@ HEADERS +=	pqi/authssl.h \
 			pqi/pqinetstatebox.h \
                         pqi/p3servicecontrol.h
 
+SOURCES += friend_server/fsclient.h \
+	   friend_server/fsbio.h \
+	   friend_server/fsmanager.h 
+
 HEADERS +=	rsserver/p3face.h \
 			rsserver/p3history.h \
 			rsserver/p3msgs.h \
@@ -569,6 +573,10 @@ SOURCES +=	pqi/authgpg.cc \
 			pqi/pqinetstatebox.cc \
                         pqi/p3servicecontrol.cc
 
+SOURCES += friend_server/fsclient.cc \
+	   friend_server/fsbio.cc \
+	   friend_server/fsmanager.cc 
+
 SOURCES += 		rsserver/p3face-config.cc \
 			rsserver/p3face-server.cc \
 			rsserver/p3face-info.cc \