dakirily commented on code in PR #453: URL: https://github.com/apache/qpid-broker-j/pull/453#discussion_r4357226513
########## broker-core/src/test/java/org/apache/qpid/server/transport/TlsTestContext.java: ########## @@ -0,0 +1,52 @@ +/* + * 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. + */ +package org.apache.qpid.server.transport; + +import java.security.KeyStore; +import java.security.cert.Certificate; + +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManagerFactory; + +import org.apache.qpid.test.utils.tls.KeyCertificatePair; +import org.apache.qpid.test.utils.tls.TlsResourceBuilder; + +final class TlsTestContext +{ + private TlsTestContext() + { Review Comment: Please provide a comment explaining private empty constructor ########## broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionWriteTest.java: ########## @@ -0,0 +1,507 @@ +/* + * 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. + */ +package org.apache.qpid.server.transport; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; + +import org.apache.qpid.server.bytebuffer.QpidByteBuffer; +import org.apache.qpid.server.logging.EventLogger; +import org.apache.qpid.server.model.Broker; +import org.apache.qpid.server.model.port.AmqpPort; +import org.apache.qpid.server.transport.network.TransportEncryption; +import org.apache.qpid.test.utils.UnitTestBase; + +class NonBlockingConnectionWriteTest extends UnitTestBase +{ + private static final int NETWORK_BUFFER_SIZE = 1024; + + private final AtomicBoolean _hasWork = new AtomicBoolean(); + private final AtomicBoolean _blocked = new AtomicBoolean(); + private final AtomicInteger _writableBytes = new AtomicInteger(); + private final List<Boolean> _blockedTransitions = new ArrayList<>(); + private ProtocolEngine _engine; + private SocketChannel _channel; + private NonBlockingConnection _connection; + + @BeforeEach + void setUp() throws Exception + { + _hasWork.set(false); + _blocked.set(false); + _writableBytes.set(0); + _blockedTransitions.clear(); + final Socket socket = mock(Socket.class); + when(socket.getRemoteSocketAddress()).thenReturn(new InetSocketAddress("localhost", 1000)); + when(socket.getLocalSocketAddress()).thenReturn(new InetSocketAddress("localhost", 5672)); + _channel = mock(SocketChannel.class); + when(_channel.socket()).thenReturn(socket); + when(_channel.write(any(ByteBuffer[].class))).thenAnswer(invocation -> + { + final ByteBuffer[] buffers = invocation.getArgument(0); + final int offset = 0; + final int length = buffers.length; + long written = 0; + for (int i = offset; i < offset + length; i++) + { + final int size = Math.min(buffers[i].remaining(), _writableBytes.get()); + buffers[i].position(buffers[i].position() + size); + _writableBytes.addAndGet(-size); + written += size; + } + return written; + }); + + _engine = mock(ProtocolEngine.class); + final AggregateTicker ticker = mock(AggregateTicker.class); + when(ticker.getTimeToNextTick(anyLong())).thenReturn(Integer.MAX_VALUE); + when(_engine.getAggregateTicker()).thenReturn(ticker); + when(_engine.processPendingIterator()).thenAnswer(invocation -> Collections.emptyIterator()); + when(_engine.hasWork()).thenAnswer(invocation -> _hasWork.get()); + when(_engine.isTransportBlockedForWriting()).thenAnswer(invocation -> _blocked.get()); + doAnswer(invocation -> + { + _hasWork.set(false); + return null; + }).when(_engine).clearWork(); + doAnswer(invocation -> + { + _hasWork.set(true); + return null; + }).when(_engine).notifyWork(); + doAnswer(invocation -> + { + final boolean blocked = invocation.getArgument(0); + if (_blocked.getAndSet(blocked) != blocked) + { + _blockedTransitions.add(blocked); + if (!blocked) + { + _engine.notifyWork(); + } + } + return null; + }).when(_engine).setTransportBlockedForWriting(anyBoolean()); + + final AmqpPort<?> port = mock(AmqpPort.class); + when(port.getNetworkBufferSize()).thenReturn(NETWORK_BUFFER_SIZE); + when(port.getContextValue(Integer.class, AmqpPort.FINAL_WRITE_THRESHOLD)).thenReturn(1); + when(port.getContextValue(Long.class, AmqpPort.FINAL_WRITE_TIMEOUT)).thenReturn(0L); + when(port.getContextValue(Integer.class, AmqpPort.MAX_GATHERING_WRITE_BUFFERS)).thenReturn(1024); + final Broker<?> broker = mock(Broker.class); + when(broker.getEventLogger()).thenReturn(mock(EventLogger.class)); + doReturn(broker).when(port).getParent(); + _connection = new NonBlockingConnection(_channel, _engine, Set.of(TransportEncryption.NONE), + () -> { }, mock(NetworkConnectionScheduler.class), port); + _connection.setSelectionTask(mock(SelectorThread.SelectionTask.class)); + } + + @AfterEach + void tearDown() + { + _writableBytes.set(Integer.MAX_VALUE); + _connection.close(); + _connection.doWork(); + } + + @Test + void blockedOutputWaitsForSocketReadiness() + { + send(128); + + assertFalse(_connection.doWork()); + + assertTrue(_connection.wantsWrite()); + assertTrue(_connection.wantsRead(), "Completed protocol work must allow incoming traffic during a blocked write"); + assertFalse(_connection.isStateChanged(), "Pending output alone must not reschedule the connection"); + assertEquals(List.of(true), _blockedTransitions); + + _engine.notifyWork(); + _connection.doWork(); + assertEquals(List.of(true), _blockedTransitions, "A failed retry must not briefly resume consumers"); + assertFalse(_connection.isStateChanged()); + + _writableBytes.set(32); + _connection.doWork(); + assertTrue(_connection.wantsWrite(), "Partial progress must retain write interest"); + assertEquals(List.of(true), _blockedTransitions); + assertFalse(_connection.isStateChanged()); + + _writableBytes.set(96); + _connection.doWork(); + assertEquals(List.of(true, false), _blockedTransitions); + assertTrue(_connection.isStateChanged(), "Unblocking must resume pending protocol output"); + assertTrue(_connection.wantsRead()); + assertFalse(_connection.wantsWrite()); + } + + @ParameterizedTest + @ValueSource(ints = {0, NETWORK_BUFFER_SIZE / 4}) + void pendingIteratorIsRescheduledWhenSecondWriteDrainsOutput(final int firstWriteBytes) throws Exception + { + final AtomicInteger completed = new AtomicInteger(); + final List<Runnable> tasks = List.of(() -> send(2 * NETWORK_BUFFER_SIZE), completed::incrementAndGet); + when(_engine.processPendingIterator()).thenReturn(tasks.iterator()); + final AtomicInteger writeCalls = new AtomicInteger(); + doAnswer(invocation -> + { + final ByteBuffer[] buffers = invocation.getArgument(0); + int writableBytes = writeCalls.getAndIncrement() == 0 ? firstWriteBytes : Integer.MAX_VALUE; + long written = 0; + for (final ByteBuffer buffer : buffers) + { + final int size = Math.min(buffer.remaining(), writableBytes); + buffer.position(buffer.position() + size); + writableBytes -= size; + written += size; + } + return written; + }).when(_channel).write(any(ByteBuffer[].class)); + + assertFalse(_connection.doWork()); + + assertEquals(2, writeCalls.get(), "The second write must drain output after the iterator is suspended"); + assertEquals(0, completed.get(), "The iterator still has runnable work after the second write"); + assertTrue(_connection.wantsRead()); + assertFalse(_connection.wantsWrite()); + assertFalse(_connection.isPartialRead()); + assertFalse(_blocked.get()); + assertTrue(_blockedTransitions.isEmpty(), "There is no backpressure transition to supply a wake-up"); + assertTrue(_connection.isStateChanged(), + "Pending protocol work must be rescheduled when output drains in the same pass"); + + assertFalse(_connection.doWork()); + assertEquals(1, completed.get(), "The retained iterator must resume exactly once"); + assertFalse(_connection.isStateChanged(), "An exhausted iterator must not keep rescheduling"); + verify(_engine).processPendingIterator(); + } + + @Test + void pendingIteratorResumesAfterOutputDrains() + { + final AtomicInteger completed = new AtomicInteger(); + final List<Runnable> tasks = List.of(() -> send(NETWORK_BUFFER_SIZE), completed::incrementAndGet); + when(_engine.processPendingIterator()).thenReturn(tasks.iterator()); + + _connection.doWork(); + + assertEquals(0, completed.get()); + assertTrue(_connection.wantsWrite()); + assertFalse(_connection.wantsRead(), "Suspended protocol work must continue to apply read backpressure"); + assertFalse(_connection.isStateChanged(), "A suspended iterator must wait for its buffered output to drain"); + + _writableBytes.set(NETWORK_BUFFER_SIZE); + _connection.doWork(); + + assertEquals(1, completed.get(), "Suspended work must resume exactly once"); + assertFalse(_connection.wantsWrite()); + } + + @ParameterizedTest + @CsvSource({"false, 768", "true, 768", "false, 1023", "true, 1023"}) + void partialWriteBelowBufferLimitResumesWorkAndReads(final boolean pendingTask, + final int remainingOutput) throws Exception Review Comment: Please fix the parameter indentation ########## broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java: ########## @@ -278,19 +279,26 @@ public boolean doWork() if (processPendingComplete) { _pendingIterator = null; - _protocolEngine.setTransportBlockedForWriting(false); + } + + if (processPendingComplete || _delegate.needsReadForWrite()) + { final boolean dataRead = doRead(); - _protocolEngine.setTransportBlockedForWriting(!doWrite()); + doWrite(); - if (!_fullyWritten || dataRead || (_delegate.needsWork() && _delegate.getNetInputBuffer().position() != 0)) + if (_fullyWritten && + (dataRead || (_delegate.needsWork() && _delegate.getNetInputBuffer().position() != 0))) Review Comment: if expression on lines 289-290 should be on one line ########## broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionSchedulingTest.java: ########## @@ -0,0 +1,455 @@ +/* + * 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. + */ +package org.apache.qpid.server.transport; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.nio.channels.SelectionKey; +import java.nio.channels.ServerSocketChannel; +import java.nio.channels.SocketChannel; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.Set; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLSocket; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; + +import org.apache.qpid.server.bytebuffer.QpidByteBuffer; +import org.apache.qpid.server.logging.EventLogger; +import org.apache.qpid.server.model.Broker; +import org.apache.qpid.server.model.port.AmqpPort; +import org.apache.qpid.server.transport.network.TransportEncryption; +import org.apache.qpid.server.util.Action; +import org.apache.qpid.test.utils.UnitTestBase; + +class NonBlockingConnectionSchedulingTest extends UnitTestBase +{ + private static final int TIMEOUT_MILLIS = 10000; + + private final List<Thread> _threads = new CopyOnWriteArrayList<>(); + private final List<TestConnection> _connections = new ArrayList<>(); + private NetworkConnectionScheduler _scheduler; + private SSLContext _sslContext; + + @BeforeAll + void createTlsContext() throws Exception + { + _sslContext = TlsTestContext.create(); + } + + @ParameterizedTest + @ValueSource(ints = {2, 4}) + void blockedSocketReturnsWorkersAndRecoversPendingOutput(final int workers) throws Exception + { + startScheduler(workers); + final TestConnection blocked = createConnection(1024); + final TestConnection independent = createConnection(1024); + final byte[] payload = new byte[2 * 1024 * 1024]; + Arrays.fill(payload, (byte) 42); + final byte[] marker = new byte[]{1, 2, 3}; + for (int i = 0; i < 3; i++) + { + blocked.submit(List.of(() -> blocked.send(payload), () -> blocked.send(marker))); + blocked.awaitWriteInterest(); + + independent.submit(List.of(() -> independent.send(marker))); + assertArrayEquals(marker, independent._client.getInputStream().readNBytes(marker.length), + "An independent connection must progress while the first socket remains blocked"); + + assertArrayEquals(payload, blocked._client.getInputStream().readNBytes(payload.length)); + assertArrayEquals(marker, blocked._client.getInputStream().readNBytes(marker.length), + "The suspended protocol iterator must resume after socket capacity becomes available"); + } + } + + @ParameterizedTest + @CsvSource({"TLSv1.2, 2", "TLSv1.2, 4", "TLSv1.3, 2", "TLSv1.3, 4"}) + void blockedTlsSocketReturnsWorkersAndRecoversPendingOutput(final String protocol, final int workers) throws Exception + { + startScheduler(workers); + final TestConnection blocked = createTlsConnection(protocol); + final TestConnection independent = createTlsConnection(protocol); + final byte[] payload = new byte[2 * 1024 * 1024]; + for (int i = 0; i < payload.length; i++) + { + payload[i] = (byte) i; + } + final byte[] marker = new byte[]{11, 22, 33}; + for (int round = 0; round < 3; round++) + { + blocked.submit(List.of(() -> blocked.send(payload), () -> blocked.send(marker))); + blocked.awaitWriteInterest(); + assertTrue(blocked._blocked.get()); + + independent.submit(List.of(() -> independent.send(marker))); + assertArrayEquals(marker, independent._client.getInputStream().readNBytes(marker.length), + "An independent TLS connection must progress while the first socket is blocked"); + + blocked._client.getOutputStream().write(42 + round); + assertEquals(Integer.valueOf(42 + round), blocked._received.poll(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS), + "TLS input must still be decrypted while encrypted output is blocked"); + blocked.awaitWriteInterest(); + assertTrue(blocked._blocked.get(), "Reading peer input must not briefly resume blocked consumers"); + + assertArrayEquals(payload, blocked._client.getInputStream().readNBytes(payload.length)); + assertArrayEquals(marker, blocked._client.getInputStream().readNBytes(marker.length), + "Pending output must resume without losing or reordering application bytes"); + blocked.awaitSelection(false); + assertFalse(blocked._blocked.get(), "Draining encrypted output must resume consumers"); + } + } + + @Test + void incomingHeartbeatsPreventReadIdleTimeoutWhileOutputIsBlocked() throws Exception + { + startScheduler(2); + final TestConnection blocked = createConnection(4 * 1024 * 1024); + blocked.submit(List.of(() -> blocked.send(new byte[2 * 1024 * 1024]))); + blocked.awaitWriteInterest(); + + final int readIdleMillis = 2000; + final AtomicLong currentTime = new AtomicLong(); + final AtomicLong lastReadTime = new AtomicLong(); + final BlockingQueue<Integer> heartbeats = new LinkedBlockingQueue<>(); + final AtomicBoolean readerIdle = new AtomicBoolean(); + when(blocked._engine.getLastReadTime()).thenAnswer(invocation -> lastReadTime.get()); + doAnswer(invocation -> + { + final QpidByteBuffer input = invocation.getArgument(0); + lastReadTime.set(currentTime.get()); + while (input.hasRemaining()) + { + heartbeats.add((int) input.get()); + } + return null; + }).when(blocked._engine).received(any()); + doAnswer(invocation -> + { + readerIdle.set(true); + blocked._connection.close(); + return null; + }).when(blocked._engine).readerIdle(); + + // Keep wall-clock scheduling timestamps out of the idle calculation so only the test advances time. + blocked._ticker.addTicker(new ServerIdleReadTimeoutTicker(mock(ServerNetworkConnection.class), + blocked._engine, readIdleMillis) + { + @Override + public int getTimeToNextTick(final long ignored) + { + return super.getTimeToNextTick(currentTime.get()); + } + }); + blocked.awaitWriteInterest(); + + for (int i = 0; i < 8; i++) + { + currentTime.addAndGet(readIdleMillis / 4); + blocked._client.getOutputStream().write(42); + assertEquals(Integer.valueOf(42), heartbeats.poll(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS), + "Incoming heartbeats must be consumed while the peer is not reading output"); + blocked.awaitWriteInterest(); + assertFalse(readerIdle.get(), "A peer sending heartbeats must not be considered idle"); + } + + assertTrue(blocked._connection.wantsWrite(), "Output must remain blocked throughout the heartbeat checks"); + currentTime.addAndGet(readIdleMillis - 1); + blocked.awaitWriteInterest(); + assertFalse(readerIdle.get(), "The connection must remain open until the read deadline"); + + currentTime.incrementAndGet(); + blocked._connection.getSelectionTask().wakeup(); + blocked._closed.get(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); + assertTrue(readerIdle.get(), "The read timeout must still close the connection after heartbeats stop"); + } + + private void startScheduler(final int workers) + { + _scheduler = new NetworkConnectionScheduler(getTestName(), 1, workers, 1L, task -> + { + final Thread thread = new Thread(task, getTestName() + "-" + _threads.size()); + _threads.add(thread); + return thread; + }); + _scheduler.start(); + } + + private TestConnection createConnection(final int networkBufferSize) throws Exception + { + final TestConnection connection = new TestConnection(_scheduler, networkBufferSize); + _connections.add(connection); + return connection; + } + + private TestConnection createTlsConnection(final String protocol) throws Exception + { + final SSLSocket client = (SSLSocket) _sslContext.getSocketFactory().createSocket(); + client.setEnabledProtocols(new String[]{protocol}); + final TestConnection connection = new TestConnection(_scheduler, 64 * 1024, client, _sslContext); + _connections.add(connection); + client.startHandshake(); + assertEquals(protocol, client.getSession().getProtocol()); + return connection; + } + + @AfterEach + void tearDown() throws Exception + { + if (_scheduler != null) + { + try + { + for (final TestConnection connection : _connections) + { + connection._connection.close(); + } + for (final TestConnection connection : _connections) + { + connection._closed.get(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); + } + } + finally + { + for (final TestConnection connection : _connections) + { + connection._client.close(); + connection._channel.close(); + } + _scheduler.close(); + try + { + for (final Thread thread : _threads) + { + thread.join(TIMEOUT_MILLIS); + assertFalse(thread.isAlive(), "Scheduler worker did not terminate: " + thread.getName()); + } + } + finally + { + for (final Thread thread : _threads) + { + thread.interrupt(); + } + for (final Thread thread : _threads) + { + thread.join(TIMEOUT_MILLIS); + } + } + } + } + } + + private static final class TestConnection + { + private final AtomicBoolean _hasWork = new AtomicBoolean(); + private final AtomicBoolean _blocked = new AtomicBoolean(); + private final AtomicReference<Iterator<Runnable>> _pending = + new AtomicReference<>(Collections.emptyIterator()); + private final AtomicReference<SelectionInterest> _selectionInterest = new AtomicReference<>(); + private final BlockingQueue<Integer> _received = new LinkedBlockingQueue<>(); + private final AggregateTicker _ticker = new AggregateTicker() + { + @Override + public void resetModified() + { + super.resetModified(); + // The selector calls this after examining a registered connection's ticker. + signalSelection(); + } + }; + private final CompletableFuture<Void> _closed = new CompletableFuture<>(); + private final Socket _client; + private final SocketChannel _channel; + private final ProtocolEngine _engine = mock(ProtocolEngine.class); + private final NonBlockingConnection _connection; + + private TestConnection(final NetworkConnectionScheduler scheduler, final int networkBufferSize) throws Exception + { + this(scheduler, networkBufferSize, new Socket(), null); + } + + private TestConnection(final NetworkConnectionScheduler scheduler, final int networkBufferSize, Review Comment: Please place each method parameter on a separate line -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
