echo/rabbitmq-tutorials
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tidy up the code for tut 2

Browse Source 
Signed-off-by: Ann <awitbrock@vmware.com>
This commit is contained in:
Ann 2011-01-24 17:15:13 +00:00
parent 8f4b1edb82
commit 914f22cd21
2 changed files with 61 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
java/NewTask.java
View File

@ -11,13 +11,10 @@ public class NewTask {
connection = factory.newConnection(); connection = factory.newConnection();
Channel channel = connection.createChannel(); Channel channel = connection.createChannel();
boolean durable = true; channel.queueDeclare("task_queue", true, false, false, null);
channel.queueDeclare("task_queue", durable, false, false, null);
String message = joinStrings(argv); String message = getMessage(argv);
if (message == "") message = "Hello World!";
// make message props persistent
channel.basicPublish( "", "task_queue", channel.basicPublish( "", "task_queue",
MessageProperties.PERSISTENT_TEXT_PLAIN, MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes()); message.getBytes());
@ -26,10 +23,17 @@ public class NewTask {
connection.close(); connection.close();
} }
private static String getMessage(String[] strings){
if (strings.length < 1)
return "Hello World!";
return joinStrings(strings);
}
private static String joinStrings(String[] strings){ private static String joinStrings(String[] strings){
String words = ""; String words = "";
for (String astring: strings) for (String astring: strings)
words = words + astring + " "; words = words.concat(astring).concat(" ");
return words.trim(); return words.trim();
} }
} }

31
java/Worker.java
View File

@ -13,35 +13,32 @@ public class Worker {
connection = factory.newConnection(); connection = factory.newConnection();
Channel channel = connection.createChannel(); Channel channel = connection.createChannel();
boolean durable = true; channel.queueDeclare("task_queue", true, false, false, null);
channel.queueDeclare("task_queue", durable, false, false, null);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
int prefetchCount = 1; channel.basicQos(1);
channel.basicQos(prefetchCount);
QueueingConsumer consumer = new QueueingConsumer(channel); QueueingConsumer consumer = new QueueingConsumer(channel);
boolean autoAck = false; channel.basicConsume("task_queue", false, consumer);
channel.basicConsume("task_queue", autoAck, consumer);
while (true) { while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery(); QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String body = new String(delivery.getBody()); String body = new String(delivery.getBody());
System.out.println(" [x] Received " + body); System.out.println(" [x] Received " + body);
Thread.sleep(doDots(body)); // simulate action
System.out.println(" [x] Done"); Thread.sleep( charCount(body, '.') * 1000);
// acknowledge System.out.println(" [x] Done" );
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
} }
} }
private static int charCount(String body, char theChar){
int count = 0;
private static int doDots(String body){ for (int index = 0; index < body.length(); index++){
// just do something to get a number for simulation if (theChar == body.charAt(index)) count++;
int x = body.indexOf('.') ; }
if (x < 0) return 0; return count;
body = body.substring(x);
return body.length() ;
} }
} }