From d3c87daf643e58d1005a18eef3539a7cef0c4bb4 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 18 Nov 2015 16:15:50 -0500 Subject: [PATCH] GH-203: Support Late Binding with RabbitMQ If the broker was down, the binding failed. Defer the queue/exchange/binding declarations until the broker is available (already supported by `RabbitAdmin`. Solves spring-cloud/spring-cloud-stream#203 Fix PubSub, Test - Too many binder prefixes on pub sub queue name. - Mismatched queue args in test case. --- .../rabbit/RabbitMessageChannelBinder.java | 159 +++++++++--------- .../binder/rabbit/RabbitBinderTests.java | 115 +++++++++++++ .../test/junit/rabbit/RabbitTestSupport.java | 105 ++++++++++++ 3 files changed, 302 insertions(+), 77 deletions(-) diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 615207799..064432040 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -30,6 +30,7 @@ import org.aopalliance.aop.Advice; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.amqp.AmqpConnectException; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.DirectExchange; @@ -42,7 +43,6 @@ import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate; -import org.springframework.amqp.rabbit.core.ChannelCallback; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.core.support.BatchingStrategy; @@ -309,6 +309,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl this.rabbitAdmin = new RabbitAdmin(connectionFactory); this.autoDeclareContext.refresh(); this.rabbitAdmin.setApplicationContext(this.autoDeclareContext); + this.rabbitAdmin.setIgnoreDeclarationExceptions(true); this.rabbitAdmin.afterPropertiesSet(); } @@ -447,17 +448,19 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl RabbitPropertiesAccessor accessor = new RabbitPropertiesAccessor(properties); String queueName = applyPrefix(accessor.getPrefix(this.defaultPrefix), name); TopicExchange exchange = new TopicExchange(queueName); - declareExchangeIfNotPresent(exchange); + declareExchange(queueName, exchange); int partitionIndex = accessor.getPartitionIndex(); + String dlqNameRoot = name; if (partitionIndex >= 0) { - queueName += "-" + partitionIndex; + String partitionSuffix = "-" + partitionIndex; + queueName += partitionSuffix; + dlqNameRoot += partitionSuffix; } - Map args = queueArgs(accessor, queueName); - Queue queue = new Queue(queueName, true, false, false, args); - declareQueueIfNotPresent(queue); - autoBindDLQ(name, accessor); + Queue queue = new Queue(queueName, true, false, false, queueArgs(accessor, queueName)); + declareQueue(queueName, queue); + autoBindDLQ(dlqNameRoot, accessor); org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with(queueName); - this.rabbitAdmin.declareBinding(binding); + declareBinding(queueName, binding); doRegisterConsumer(name, moduleInputChannel, queue, accessor, false); bindExistingProducerDirectlyIfPossible(name, moduleInputChannel); } @@ -473,28 +476,18 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl validateConsumerProperties(name, properties, SUPPORTED_PUBSUB_CONSUMER_PROPERTIES); String prefix = accessor.getPrefix(this.defaultPrefix); TopicExchange exchange = new TopicExchange(applyPrefix(prefix, exchangeName)); - declareExchangeIfNotPresent(exchange); + declareExchange(exchange.getName(), exchange); Queue queue; boolean durable = accessor.isDurable(this.defaultDurableSubscription); String queueName = applyPrefix(prefix, name); if (durable) { - Map args = queueArgs(accessor, queueName); - queue = new Queue(queueName, true, false, false, args); + queue = new Queue(queueName, true, false, false, queueArgs(accessor, queueName)); } else { queue = new Queue(queueName, false, false, true); } - declareQueueIfNotPresent(queue); - org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with("#"); - this.rabbitAdmin.declareBinding(binding); - // register with context so they will be redeclared after a connection failure if auto-delete - if (!durable && !this.autoDeclareContext.containsBean(name)) { - this.autoDeclareContext.getBeanFactory().registerSingleton(name, queue); - } - String bindingBeanName = exchange.getName() + "." + queue.getName() + ".binding"; - if (!durable && !this.autoDeclareContext.containsBean(bindingBeanName)) { - this.autoDeclareContext.getBeanFactory().registerSingleton(bindingBeanName, binding); - } + declareQueue(queueName, queue); + declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with("#")); doRegisterConsumer(name, moduleInputChannel, queue, accessor, true); if (durable) { autoBindDLQ(name, accessor); @@ -586,7 +579,6 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl RepublishMessageRecoverer republishMessageRecoverer = new RepublishMessageRecoverer(errorTemplate, deadLetterExchangeName(prefix), applyPrefix(prefix, name)); - // TODO: Add container id to republished message headers? (Needs AMQP-489). return republishMessageRecoverer; } else { @@ -621,26 +613,27 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl String partitionKeyExtractorClass = properties.getPartitionKeyExtractorClass(); Expression partitionKeyExpression = properties.getPartitionKeyExpression(); TopicExchange exchange = new TopicExchange(queueName); - declareExchangeIfNotPresent(exchange); + declareExchange(queueName, exchange); AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(rabbitTemplate); endpoint.setExchangeName(exchange.getName()); if (partitionKeyExpression == null && !StringUtils.hasText(partitionKeyExtractorClass)) { - Queue queue = new Queue(queueName); - declareQueueIfNotPresent(queue); + Queue queue = new Queue(queueName, true, false, false, queueArgs(properties, queueName)); + declareQueue(name, queue); + autoBindDLQ(name, properties); endpoint.setRoutingKey(queueName); org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with(queueName); - this.rabbitAdmin.declareBinding(binding); + declareBinding(queueName, binding); } else { endpoint.setExpressionRoutingKey(EXPRESSION_PARSER.parseExpression(buildPartitionRoutingExpression (queueName))); // if the stream is partitioned, create one queue for each target partition for (int i = 0; i < properties.getNextModuleCount(); i++) { - Queue queue = new Queue(queueName + "-" + i); - this.rabbitAdmin.declareQueue(queue); - org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange) - .with(queue.getName()); - this.rabbitAdmin.declareBinding(binding); + String partitionSuffix = "-" + i; + Queue queue = new Queue(queueName + partitionSuffix, true, false, false, queueArgs(properties, queueName)); + declareQueue(queue.getName(), queue); + autoBindDLQ(name + partitionSuffix, properties); + declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(queue.getName())); } } configureOutboundHandler(endpoint, properties); @@ -663,7 +656,8 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl validateProducerProperties(name, properties, SUPPORTED_PUBSUB_PRODUCER_PROPERTIES); RabbitPropertiesAccessor accessor = new RabbitPropertiesAccessor(properties); String exchangeName = applyPrefix(accessor.getPrefix(this.defaultPrefix), name); - declareExchangeIfNotPresent(new TopicExchange(exchangeName)); + TopicExchange exchange = new TopicExchange(exchangeName); + declareExchange(exchangeName, exchange); AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(determineRabbitTemplate(accessor)); endpoint.setExchangeName(exchangeName); endpoint.setRoutingKey(name); @@ -731,11 +725,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl + this.getIdGenerator().generateId(); this.doRegisterProducer(name, requests, queue, replyQueueName, accessor); Queue replyQueue = new Queue(replyQueueName, false, false, true); // auto-delete - declareQueueIfNotPresent(replyQueue); - // register with context so it will be redeclared after a connection failure - if (!this.autoDeclareContext.containsBean(replyQueueName)) { - this.autoDeclareContext.getBeanFactory().registerSingleton(replyQueueName, replyQueue); - } + declareQueue(replyQueueName, replyQueue); this.doRegisterConsumer(name, replies, replyQueue, accessor, false); } @@ -748,7 +738,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl validateConsumerProperties(name, properties, SUPPORTED_REPLYING_CONSUMER_PROPERTIES); RabbitPropertiesAccessor accessor = new RabbitPropertiesAccessor(properties); Queue requestQueue = new Queue(applyPrefix(accessor.getPrefix(this.defaultPrefix), applyRequests(name))); - declareQueueIfNotPresent(requestQueue); + declareQueue(requestQueue.getName(), requestQueue); this.doRegisterConsumer(name, requests, requestQueue, accessor, false); AmqpOutboundEndpoint replyQueue = new AmqpOutboundEndpoint(this.rabbitTemplate); @@ -758,37 +748,6 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl doRegisterProducer(name, replies, replyQueue, accessor); } - /** - * Try passive declaration first, in case the user has pre-configured the queue with incompatible arguments. - * @param queue The queue. - */ - private void declareQueueIfNotPresent(Queue queue) { - if (this.rabbitAdmin.getQueueProperties(queue.getName()) == null) { - this.rabbitAdmin.declareQueue(queue); - } - } - - /** - * Try passive declaration first, in case the user has pre-configured the exchange with incompatible arguments. - * @param exchange - */ - private void declareExchangeIfNotPresent(final Exchange exchange) { - this.rabbitTemplate.execute(new ChannelCallback() { - - @Override - public Void doInRabbit(Channel channel) throws Exception { - try { - channel.exchangeDeclarePassive(exchange.getName()); - } - catch (IOException e) { - RabbitMessageChannelBinder.this.rabbitAdmin.declareExchange(exchange); - } - return null; - } - - }); - } - /** * If so requested, declare the DLX/DLQ and bind it. The DLQ is bound to the DLX with a routing key of the original * queue name because we use default exchange routing by queue name for the original message. @@ -805,14 +764,50 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl String queueName = applyPrefix(prefix, name); String dlqName = constructDLQName(queueName); Queue dlq = new Queue(dlqName); - declareQueueIfNotPresent(dlq); + declareQueue(dlqName, dlq); final String dlxName = deadLetterExchangeName(prefix); final DirectExchange dlx = new DirectExchange(dlxName); - declareExchangeIfNotPresent(dlx); - this.rabbitAdmin.declareBinding(BindingBuilder.bind(dlq).to(dlx).with(queueName)); + declareExchange(dlxName, dlx); + declareBinding(dlqName, BindingBuilder.bind(dlq).to(dlx).with(queueName)); } } + public void declareQueue(String beanName, Queue queue) { + try { + this.rabbitAdmin.declareQueue(queue); + } + catch (AmqpConnectException e) { + if (logger.isDebugEnabled()) { + logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available"); + } + } + addToAutoDeclareContext(beanName, queue); + } + + public void declareExchange(final String rootName, final Exchange exchange) { + try { + this.rabbitAdmin.declareExchange(exchange); + } + catch (AmqpConnectException e) { + if (logger.isDebugEnabled()) { + logger.debug("Declaration of exchange: " + exchange.getName() + " deferred - connection not available"); + } + } + addToAutoDeclareContext(rootName + ".exchange", exchange); + } + + public void declareBinding(String rootName, org.springframework.amqp.core.Binding binding) { + try { + this.rabbitAdmin.declareBinding(binding); + } + catch (AmqpConnectException e) { + if (logger.isDebugEnabled()) { + logger.debug("Declaration of binding: " + rootName + ".binding deferred - connection not available"); + } + } + addToAutoDeclareContext(rootName + ".binding", binding); + } + private String deadLetterExchangeName(String prefix) { return prefix + DEAD_LETTER_EXCHANGE; } @@ -835,11 +830,21 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl cleanAutoDeclareContext(BinderUtils.groupedName(name, group)); } + private void addToAutoDeclareContext(String name, Object bean) { + synchronized (this.autoDeclareContext) { + if (!this.autoDeclareContext.containsBean(name)) { + this.autoDeclareContext.getBeanFactory().registerSingleton(name, bean); + } + } + } + private void cleanAutoDeclareContext(String name) { - if (this.autoDeclareContext.containsBean(name)) { - ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext.getBeanFactory(); - if (beanFactory instanceof DefaultListableBeanFactory) { - ((DefaultListableBeanFactory) beanFactory).destroySingleton(name); + synchronized(this.autoDeclareContext) { + if (this.autoDeclareContext.containsBean(name)) { + ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext.getBeanFactory(); + if (beanFactory instanceof DefaultListableBeanFactory) { + ((DefaultListableBeanFactory) beanFactory).destroySingleton(name); + } } } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index e08232fdb..dbcc5f225 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -27,6 +27,7 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -49,6 +50,7 @@ import org.mockito.ArgumentCaptor; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; @@ -56,12 +58,14 @@ import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor; import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderProperties; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.PartitionCapableBinderTests; import org.springframework.cloud.stream.binder.Spy; import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport; +import org.springframework.context.support.AbstractApplicationContext; import org.springframework.expression.spel.standard.SpelExpression; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; @@ -529,6 +533,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); Map args = new HashMap(); args.put("x-dead-letter-exchange", TEST_PREFIX + "DLX"); + args.put("x-dead-letter-routing-key", TEST_PREFIX + "dlqpubtest"); Queue queue = new Queue(TEST_PREFIX + "dlqpubtest", true, false, false, args); admin.declareQueue(queue); @@ -629,6 +634,116 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { binder.unbindConsumers("batching.0"); } + /* + * Test late binding due to broker down; queues with and without DLQs, and + * partitioned queues. + */ + @Test + public void testLateBinding() throws Exception { + RabbitTestSupport.RabbitProxy proxy = new RabbitTestSupport.RabbitProxy(); + CachingConnectionFactory cf = new CachingConnectionFactory("localhost", proxy.getPort()); + RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(cf); + AbstractApplicationContext applicationContext = mock(AbstractApplicationContext.class); + when(applicationContext.getBeanFactory()).thenReturn(mock(ConfigurableListableBeanFactory.class)); + binder.setApplicationContext(applicationContext); + binder.setDefaultAutoBindDLQ(true); + binder.afterPropertiesSet(); + Properties properties = new Properties(); + properties.put("prefix", "latebinder."); + + MessageChannel moduleOutputChannel = new DirectChannel(); + binder.bindProducer("late.0", moduleOutputChannel, properties); + + QueueChannel moduleInputChannel = new QueueChannel(); + binder.bindConsumer("late.0", moduleInputChannel, properties); + + properties.put("partitionKeyExpression", "payload.equals('0') ? 0 : 1"); + properties.put("partitionSelectorExpression", "hashCode()"); + properties.put("nextModuleCount", "2"); + + MessageChannel partOutputChannel = new DirectChannel(); + binder.bindProducer("partlate.0", partOutputChannel, properties); + + QueueChannel partInputChannel0 = new QueueChannel(); + QueueChannel partInputChannel1 = new QueueChannel(); + properties.clear(); + properties.put("prefix", "latebinder."); + properties.put("partitionIndex", "0"); + binder.bindConsumer("partlate.0", partInputChannel0, properties); + properties.put("partitionIndex", "1"); + binder.bindConsumer("partlate.0", partInputChannel1, properties); + + binder.setDefaultAutoBindDLQ(false); + properties.clear(); + properties.put("prefix", "latebinder."); + MessageChannel noDLQOutputChannel = new DirectChannel(); + binder.bindProducer("lateNoDLQ.0", noDLQOutputChannel, properties); + + QueueChannel noDLQInputChannel = new QueueChannel(); + binder.bindConsumer("lateNoDLQ.0", noDLQInputChannel, properties); + + MessageChannel pubSubOutputChannel = new DirectChannel(); + binder.bindPubSubProducer("latePubSub", pubSubOutputChannel, properties); + QueueChannel pubSubInputChannel = new QueueChannel(); + binder.bindPubSubConsumer("latePubSub", pubSubInputChannel, "lategroup", properties); + QueueChannel durablePubSubInputChannel = new QueueChannel(); + properties.setProperty("durableSubscription", "true"); + binder.bindPubSubConsumer("latePubSub", durablePubSubInputChannel, "lateDurableGroup", properties); + + proxy.start(); + + moduleOutputChannel.send(new GenericMessage<>("foo")); + Message message = moduleInputChannel.receive(10000); + assertNotNull(message); + assertEquals("foo", message.getPayload()); + + noDLQOutputChannel.send(new GenericMessage<>("bar")); + message = noDLQInputChannel.receive(10000); + assertNotNull(message); + assertEquals("bar", message.getPayload()); + + pubSubOutputChannel.send(new GenericMessage<>("baz")); + message = pubSubInputChannel.receive(10000); + assertNotNull(message); + assertEquals("baz", message.getPayload()); + message = durablePubSubInputChannel.receive(10000); + assertNotNull(message); + assertEquals("baz", message.getPayload()); + + partOutputChannel.send(new GenericMessage<>("0")); + partOutputChannel.send(new GenericMessage<>("1")); + message = partInputChannel0.receive(10000); + assertNotNull(message); + assertEquals("0", message.getPayload()); + message = partInputChannel1.receive(10000); + assertNotNull(message); + assertEquals("1", message.getPayload()); + + binder.unbindProducer("late.0", moduleOutputChannel); + binder.unbindConsumer("late.0", moduleInputChannel); + binder.unbindProducer("partlate.0", moduleOutputChannel); + binder.unbindConsumers("partlate.0"); + + proxy.stop(); + cf.destroy(); + + RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); + admin.deleteQueue("latebinder.late.0"); + admin.deleteQueue("latebinder.lateNoDLQ.0"); + admin.deleteQueue("latebinder.partlate.0-0"); + admin.deleteQueue("latebinder.partlate.0-1"); + admin.deleteQueue("latebinder.late.0.dlq"); + admin.deleteQueue("latebinder.partlate.0-0.dlq"); + admin.deleteQueue("latebinder.partlate.0-1.dlq"); + admin.deleteQueue("latebinder.lateDurableGroup.latePubSub"); + admin.deleteExchange("latebinder.late.0"); + admin.deleteExchange("latebinder.lateNoDLQ.0"); + admin.deleteExchange("latebinder.partlate.0"); + admin.deleteExchange("latebinder.latePubSub"); + admin.deleteExchange("latebinder.DLX"); + this.rabbitAvailableRule.getResource().destroy(); + } + private SimpleMessageListenerContainer verifyContainer(AbstractEndpoint endpoint) { SimpleMessageListenerContainer container; Advice retry; diff --git a/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/rabbit/RabbitTestSupport.java b/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/rabbit/RabbitTestSupport.java index 84f597dc1..1de2403c8 100644 --- a/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/rabbit/RabbitTestSupport.java +++ b/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/rabbit/RabbitTestSupport.java @@ -16,8 +16,15 @@ package org.springframework.cloud.stream.test.junit.rabbit; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.ServerSocket; import java.net.Socket; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import javax.net.ServerSocketFactory; import javax.net.SocketFactory; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; @@ -58,4 +65,102 @@ public class RabbitTestSupport extends AbstractExternalResourceTestSupport= 0) { + os.write(c); + } + } + catch (IOException e) { + try { + socket.close(); + rabbitSocket.close(); + } + catch (IOException e1) { + } + } + } + }); + InputStream is = socket.getInputStream(); + OutputStream os = rabbitSocket.getOutputStream(); + int c; + while ((c = is.read()) >= 0) { + os.write(c); + } + } + catch (IOException e) { + try { + socket.close(); + } + catch (IOException e1) { + } + } + } + + }); + } + } + catch (IOException e) { + try { + serverSocket.close(); + } + catch (IOException e1) { + } + } + } + }); + } + + public void stop() throws IOException { + this.serverSocket.close(); + } + + } + }