// Copyright (C) 2002 Dominic Letourneau #include "Vector.h" #include "BufferedNode.h" #include "ObjectRef.h" #include "ObjectParser.h" #include #include "Stream.h" #include "SocketStream.h" using namespace std; namespace FD { class BroadcastLoad; DECLARE_NODE(BroadcastLoad) /*Node * * @name BroadcastLoad * @category IO * @description Load an object from file (registered type) * * @input_name SOCKET * @input_description The stream we are loading from * @input_type Stream * * @output_name OUTPUT * @output_description The loaded object * END*/ class BroadcastLoad : public BufferedNode { protected: /**The ID of the 'output' output*/ int outputID; /**The ID of the 'stream' input*/ int socketInputID; public: BroadcastLoad(string nodeName, ParameterSet params) : BufferedNode(nodeName, params) { outputID = addOutput("OUTPUT"); socketInputID = addInput("SOCKET"); inOrder = true; } void calculate(int output_id, int count, Buffer &out) { istream & my_stream = (istream&) object_cast(getInput(socketInputID,count)); socket_iostream *my_socket_iostream = dynamic_cast(&my_stream); if (!my_socket_iostream) { throw new GeneralException("Invalid socket",__FILE__,__LINE__); } //getting socket socket_streambuf &my_streambuf = (socket_streambuf&) (*my_socket_iostream); Vector *output_vect = new Vector; int object_count = 0; while(1) { //32 kb of data unsigned char packet[1024 * 32]; memset(&packet[0],0,1024 * 32); int size; size = my_streambuf.recv_packet(&packet[0], 1024 * 32); if (size > 0) { //cerr<<"BROADCAST LOAD SIZE : "<> my_object; output_vect->push_back(my_object); object_count++; } catch(BaseException *e) { cerr<<"BroadcastLoad (exception)"<print(cerr); delete e; } } else { break; } } //cerr<<"BroadcastLoad read : "<