You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2006/11/21 12:49:15 UTC
svn commit: r477636 - in /incubator/qpid/trunk/qpid/cpp:
src/qpid/client/Connector.cpp src/qpid/client/Connector.h
test/client/echo_service.cpp
Author: gsim
Date: Tue Nov 21 03:49:14 2006
New Revision: 477636
URL: http://svn.apache.org/viewvc?view=rev&rev=477636
Log:
Added another example for client api.
Added:
incubator/qpid/trunk/qpid/cpp/test/client/echo_service.cpp (with props)
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.h
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp?view=diff&rev=477636&r1=477635&r2=477636
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp Tue Nov 21 03:49:14 2006
@@ -98,6 +98,12 @@
}
}
+void Connector::handleClosed(){
+ closed = true;
+ socket.close();
+ if(shutdownHandler) shutdownHandler->shutdown();
+}
+
void Connector::checkIdle(ssize_t status){
if(timeoutHandler){
int64_t now = Time::now().msecs();
@@ -106,9 +112,7 @@
timeoutHandler->idleIn();
}
}else if(status == Socket::SOCKET_EOF){
- closed = true;
- socket.close();
- if(shutdownHandler) shutdownHandler->shutdown();
+ handleClosed();
}else{
lastIn = now;
}
@@ -153,7 +157,7 @@
ssize_t received = socket.recv(inbuf.start(), available);
checkIdle(received);
- if(received > 0){
+ if(!closed && received > 0){
inbuf.move(received);
inbuf.flip();//position = 0, limit = total data read
@@ -168,5 +172,6 @@
}
}catch(QpidError error){
std::cout << "Error [" << error.code << "] " << error.msg << " (" << error.file << ":" << error.line << ")" << std::endl;
+ handleClosed();
}
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.h?view=diff&rev=477636&r1=477635&r2=477636
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.h Tue Nov 21 03:49:14 2006
@@ -70,6 +70,7 @@
void setSocketTimeout();
void run();
+ void handleClosed();
public:
Connector(bool debug = false, u_int32_t buffer_size = 1024);
Added: incubator/qpid/trunk/qpid/cpp/test/client/echo_service.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/test/client/echo_service.cpp?view=auto&rev=477636
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/test/client/echo_service.cpp (added)
+++ incubator/qpid/trunk/qpid/cpp/test/client/echo_service.cpp Tue Nov 21 03:49:14 2006
@@ -0,0 +1,197 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include <qpid/QpidError.h>
+#include <qpid/client/Channel.h>
+#include <qpid/client/Connection.h>
+#include <qpid/client/Exchange.h>
+#include <qpid/client/MessageListener.h>
+#include <qpid/client/Queue.h>
+#include <qpid/sys/Time.h>
+#include <iostream>
+#include <sstream>
+
+using namespace qpid::client;
+using namespace qpid::sys;
+
+class EchoServer : public MessageListener{
+ Channel* const channel;
+public:
+ EchoServer(Channel* channel);
+ virtual void received(Message& msg);
+};
+
+class LoggingListener : public MessageListener{
+public:
+ virtual void received(Message& msg);
+};
+
+class Args{
+ string host;
+ int port;
+ bool trace;
+ bool help;
+ bool client;
+public:
+ inline Args() : host("localhost"), port(5672), trace(false), help(false), client(false){}
+ void parse(int argc, char** argv);
+ void usage();
+
+ inline const string& getHost() const { return host;}
+ inline int getPort() const { return port; }
+ inline bool getTrace() const { return trace; }
+ inline bool getHelp() const { return help; }
+ inline bool getClient() const { return client; }
+};
+
+int main(int argc, char** argv){
+ const std::string echo_service("echo_service");
+ Args args;
+ args.parse(argc, argv);
+ if (args.getHelp()) {
+ args.usage();
+ } else if (args.getClient()) {
+ try {
+ //Create connection & open a channel
+ Connection connection(args.getTrace());
+ connection.open(args.getHost(), args.getPort());
+ Channel channel;
+ connection.openChannel(&channel);
+
+ //Setup: declare the private 'response' queue and bind it
+ //to the direct exchange by its name which will be
+ //generated by the server
+ Queue response;
+ channel.declareQueue(response);
+ qpid::framing::FieldTable emptyArgs;
+ channel.bind(Exchange::DEFAULT_DIRECT_EXCHANGE, response, response.getName(), emptyArgs);
+
+ //Consume from the response queue, logging all echoed message to console:
+ LoggingListener listener;
+ std::string tag;
+ channel.consume(response, tag, &listener);
+
+ //Process incoming requests on a new thread
+ channel.start();
+
+ //get messages from console and send them:
+ std::string text;
+ std::cout << "Enter text to send:" << std::endl;
+ while (std::getline(std::cin, text)) {
+ std::cout << "Sending " << text << " to echo server." << std::endl;
+ Message msg;
+ msg.getHeaders().setString("RESPONSE_QUEUE", response.getName());
+ msg.setData(text);
+ channel.publish(msg, Exchange::DEFAULT_DIRECT_EXCHANGE, echo_service);
+
+ std::cout << "Enter text to send:" << std::endl;
+ }
+
+ connection.close();
+ } catch(qpid::QpidError error) {
+ std::cout << "Error [" << error.code << "] " << error.msg << " (" << error.file << ":" << error.line << ")" << std::endl;
+ }
+ } else {
+ try {
+ //Create connection & open a channel
+ Connection connection(args.getTrace());
+ connection.open(args.getHost(), args.getPort());
+ Channel channel;
+ connection.openChannel(&channel);
+
+ //Setup: declare the 'request' queue and bind it to the direct exchange with a 'well known' name
+ Queue request("request");
+ channel.declareQueue(request);
+ qpid::framing::FieldTable emptyArgs;
+ channel.bind(Exchange::DEFAULT_DIRECT_EXCHANGE, request, echo_service, emptyArgs);
+
+ //Consume from the request queue, echoing back all messages received to the client that sent them
+ EchoServer server(&channel);
+ std::string tag = "server_tag";
+ channel.consume(request, tag, &server);
+
+ //Process incoming requests on the main thread
+ channel.run();
+
+ connection.close();
+ } catch(qpid::QpidError error) {
+ std::cout << "Error [" << error.code << "] " << error.msg << " (" << error.file << ":" << error.line << ")" << std::endl;
+ }
+ }
+}
+
+EchoServer::EchoServer(Channel* _channel) : channel(_channel){}
+
+void EchoServer::received(Message& message)
+{
+ //get name of response queues binding to the default direct exchange:
+ const std::string name = message.getHeaders().getString("RESPONSE_QUEUE");
+
+ if (name.empty()) {
+ std::cout << "Cannot echo " << message.getData() << ", no response queue specified." << std::endl;
+ } else {
+ //print message to console:
+ std::cout << "Echoing " << message.getData() << " back to " << name << std::endl;
+
+ //'echo' the message back:
+ channel->publish(message, Exchange::DEFAULT_DIRECT_EXCHANGE, name);
+ }
+}
+
+void LoggingListener::received(Message& message)
+{
+ //print message to console:
+ std::cout << "Received echo: " << message.getData() << std::endl;
+}
+
+
+void Args::parse(int argc, char** argv){
+ for(int i = 1; i < argc; i++){
+ string name(argv[i]);
+ if("-help" == name){
+ help = true;
+ break;
+ }else if("-host" == name){
+ host = argv[++i];
+ }else if("-port" == name){
+ port = atoi(argv[++i]);
+ }else if("-trace" == name){
+ trace = true;
+ }else if("-client" == name){
+ client = true;
+ }else{
+ std::cout << "Warning: unrecognised option " << name << std::endl;
+ }
+ }
+}
+
+void Args::usage(){
+ std::cout << "Options:" << std::endl;
+ std::cout << " -help" << std::endl;
+ std::cout << " Prints this usage message" << std::endl;
+ std::cout << " -host <host>" << std::endl;
+ std::cout << " Specifies host to connect to (default is localhost)" << std::endl;
+ std::cout << " -port <port>" << std::endl;
+ std::cout << " Specifies port to conect to (default is 5762)" << std::endl;
+ std::cout << " -trace" << std::endl;
+ std::cout << " Indicates that the frames sent and received should be logged" << std::endl;
+ std::cout << " -client" << std::endl;
+ std::cout << " Run as a client (else will run as a server)" << std::endl;
+}
Propchange: incubator/qpid/trunk/qpid/cpp/test/client/echo_service.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/test/client/echo_service.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date