andi/perfdb
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Activity

extract utility method

Browse Source 
a method to send json over tcp can be used by several tests
This commit is contained in:
Andreas Huber
2017-04-08 08:21:50 +02:00
parent 81b7cfa7bd
commit 72436e9c8c
3 changed files with 85 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
pdb-ui/src/test/java/org/lucares/performance/db/ingestor/PdbTestUtil.java Normal file
View File

@@ -0,0 +1,80 @@
package org.lucares.performance.db.ingestor;
import java.io.IOException;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import org.lucares.pdbui.TcpIngestor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.databind.ObjectMapper;
public class PdbTestUtil {
private static final Logger LOGGER = LoggerFactory.getLogger(PdbTestUtil.class);
private static final Map<String, Object> POISON = new HashMap<>();
@SafeVarargs
public static final void send(final Map<String, Object>... entries) throws IOException, InterruptedException {
final LinkedBlockingDeque<Map<String, Object>> queue = new LinkedBlockingDeque<>(Arrays.asList(entries));
queue.put(POISON);
send(queue);
}
public static final void send(final BlockingQueue<Map<String, Object>> aEntriesSupplier) throws IOException {
final ObjectMapper mapper = new ObjectMapper();
final SocketChannel channel = connect();
Map<String, Object> entry;
while ((entry = aEntriesSupplier.poll()) != POISON) {
final StringBuilder streamData = new StringBuilder();
streamData.append(mapper.writeValueAsString(entry));
streamData.append("\n");
final ByteBuffer src = ByteBuffer.wrap(streamData.toString().getBytes(StandardCharsets.UTF_8));
channel.write(src);
}
try {
// ugly workaround: the channel was closed too early and not all
// data was received
TimeUnit.MILLISECONDS.sleep(10);
} catch (final InterruptedException e) {
throw new IllegalStateException(e);
}
channel.close();
LOGGER.trace("closed sender connection");
}
private static SocketChannel connect() throws IOException {
SocketChannel result = null;
while (true) {
try {
result = SocketChannel.open();
result.configureBlocking(true);
result.connect(new InetSocketAddress("127.0.0.1", TcpIngestor.PORT));
break;
} catch (final ConnectException e) {
// server socket not yet ready, it should be ready any time soon
}
}
return result;
}
}

57
pdb-ui/src/test/java/org/lucares/performance/db/ingestor/TcpIngestorTest.java
View File

@@ -1,11 +1,6 @@
package org.lucares.performance.db.ingestor;
import java.io.IOException;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Instant;
@@ -16,7 +11,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.lucares.pdb.api.Entry;
import org.lucares.pdbui.TcpIngestor;
@@ -29,8 +23,6 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
import com.fasterxml.jackson.databind.ObjectMapper;
import liquibase.exception.LiquibaseException;
@Test
@@ -72,7 +64,7 @@ public class TcpIngestorTest {
entryB.put("host", host);
entryB.put("tags", Collections.emptyList());
send(entryA, entryB);
PdbTestUtil.send(entryA, entryB);
} catch (final Exception e) {
LOGGER.error("", e);
throw e;
@@ -90,51 +82,6 @@ public class TcpIngestorTest {
}
}
@SafeVarargs
private final void send(final Map<String, Object>... entries) throws IOException {
final StringBuilder streamData = new StringBuilder();
final ObjectMapper mapper = new ObjectMapper();
for (final Map<String, Object> entry : entries) {
streamData.append(mapper.writeValueAsString(entry));
streamData.append("\n");
}
final SocketChannel channel = connect();
final ByteBuffer src = ByteBuffer.wrap(streamData.toString().getBytes(StandardCharsets.UTF_8));
channel.write(src);
try {
// ugly workaround: the channel was close too early and not all data
// was received
TimeUnit.MILLISECONDS.sleep(10);
} catch (final InterruptedException e) {
throw new IllegalStateException(e);
}
channel.close();
LOGGER.trace("closed sender connection");
}
private SocketChannel connect() throws IOException {
SocketChannel result = null;
while (true) {
try {
result = SocketChannel.open();
result.configureBlocking(true);
result.connect(new InetSocketAddress("127.0.0.1", TcpIngestor.PORT));
break;
} catch (final ConnectException e) {
// server socket not yet ready, it should be ready any time soon
}
}
return result;
}
@Test
public void testIngestionThreadDoesNotDieOnErrors() throws Exception {
final OffsetDateTime invalidDate = OffsetDateTime.ofInstant(Instant.ofEpochMilli(-1), ZoneOffset.UTC);
@@ -157,7 +104,7 @@ public class TcpIngestorTest {
entryB.put("host", host);
entryB.put("tags", Collections.emptyList());
send(entryA, entryB);
PdbTestUtil.send(entryA, entryB);
}
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {