RCF Subscription auto reconnect

[wmz5858]

i'm working currently with version RCF rcf_2.2.0.0
My server updates my clients using publish/subscribe.

if the rcf publish crash or restart，i want the subscribe auto reconnect to the publish.
so i create Subscription and call setOnSubscriptionDisconnect bind a callback.
on the callback function i create a new Subscription object, but i cannot get any message from server.
what's the problem?


-----------------------client test----------------------------------------

RCF_BEGIN(I_PrintService, "I_PrintService")
RCF_METHOD_V1(void, Print, const std::string &)
RCF_END(I_PrintService)

class PrintService
{
public:
void Print(const std::string & s)
{
std::cout << "msg: " << s << std::endl;
}
};



void onSubscriptionDisconnected(RCF::RcfSession& session, RCF::SubscriptionPtr& ptr, RCF::RcfServer* pSrv, PrintService* pPrinter, const RCF::TcpEndpoint* pEndPoint)
{
// Handle subscription disconnection here.
if (ptr != nullptr)
{
printf("close old ptr \r\n");
ptr->close();
}

try
{
RCF::SubscriptionParms subParms;
//subParms.setTopicName("123456789");
subParms.setPublisherEndpoint(*pEndPoint);
subParms.setOnSubscriptionDisconnect(boost::bind(&onSubscriptionDisconnected, boost::placeholders::_1, ptr, pSrv, pPrinter, pEndPoint));
ptr = pSrv->createSubscription<I_PrintService>(*pPrinter,subParms);

printf("success create Subscription again {%s}\r\n", ptr->getRcfSessionPtr()->isConnected()?"connected":"not connected");
}

catch (const RCF::Exception & e)
{
std::cout << "Error: " << e.getErrorString() << std::endl;

RCF::ThreadPoolPtr const pThrdPool = pSrv->getThreadPool();
RCF::AsioIoService* pIoSrv = pThrdPool->getIoService();
::Sleep(1000 * 5);

pIoSrv->post(boost::bind(&onSubscriptionDisconnected, boost::ref(*((RCF::RcfSession*)0)), ptr, pSrv, pPrinter, pEndPoint));
}

}


int main()
{

RCF::RcfInitDeinit g;
const RCF::TcpEndpoint g_srvAddr("127.0.0.1", 50001);
PrintService helloWorld;
RCF::RcfServer subscriptionServer((RCF::TcpEndpoint(-1)));
subscriptionServer.start();
RCF::SubscriptionPtr ptrSubscription;
onSubscriptionDisconnected(*((RCF::RcfSession*)0), ptrSubscription, &subscriptionServer, &helloWorld, &g_srvAddr);
getchar();

if (ptrSubscription != nullptr)
ptrSubscription->close();
return 0;
}

-----------------------client test----------------------------------------




-----------------------server test----------------------------------------
RCF_BEGIN(I_PrintService, "I_PrintService")
RCF_METHOD_V1(void, Print, const std::string &)
RCF_END(I_PrintService)
// Server implementation of the I_PrintService RCF interface.
class PrintService
{
public:
void Print(const std::string & s)
{
std::cout << "I_PrintService service: " << s << std::endl;
}
};


bool onSubscriberConnect(RCF::RcfSession & session, const std::string & topicName)
{
// Return true to allow access, false otherwise.
// ...
std::cout <<__FUNCTION__<< topicName << std::endl;
return true;
}

void onSubscriberDisconnect(RCF::RcfSession & session, const std::string & topicName)
{
// ...
std::cout << __FUNCTION__ << topicName << std::endl;
}

int main()
{
RCF::RcfInitDeinit g;
try
{
RCF::RcfServer server(RCF::TcpEndpoint("0.0.0.0", 50001));
server.start();
RCF::PublisherParms pubParms;
pubParms.setOnSubscriberConnect(onSubscriberConnect);
pubParms.setOnSubscriberDisconnect(onSubscriberDisconnect);

{
typedef boost::shared_ptr< RCF::Publisher<I_PrintService> > HelloWorldPublisherPtr;
HelloWorldPublisherPtr pubPtr = server.createPublisher<I_PrintService>(pubParms);
for (int i = 0; i < 10000; ++i)
{
::Sleep(1000 * 3);
pubPtr->publish().Print("Hello World--------->");
}
pubPtr->close();
}


std::cin.get();
}
catch (const RCF::Exception & e)
{
std::cout << "Error: " << e.getErrorString() << std::endl;
}

getchar();
return 0;
}
-----------------------server test----------------------------------------

[jarl]

Hi,

It looks like there is a problem with how you are binding parameters with boost::bind. When you bind like this:

Code: Select all

boost::bind(&onSubscriptionDisconnected, boost::placeholders::_1, ptr, pSrv, pPrinter, pEndPoint)

, onSubscriptionDisconnected() will get a copy of ptr, not a reference to it. If you want to pass a reference you need to use boost::ref, like this:

Code: Select all

boost::bind(&onSubscriptionDisconnected, boost::placeholders::_1, boost::ref(ptr), pSrv, pPrinter, pEndPoint)

[wmz5858]

Thank you for the help
That’s the problem.