wire proxy - sboesebeck/morphium GitHub Wiki
Morphium's test sources ship a small, reusable TCP proxy for the MongoDB wire protocol:
WireProxy. It sits between any wire-protocol client (Morphium, the official drivers,
mongosh) and any wire-protocol backend (MongoDB or PoppyDB) and gives you three things
that are otherwise hard to get in a test:
-
Fault injection — freeze, reset, or cleanly close connections at runtime, without
touching the server process. This is how
DriverFailoverProxyTestreproduces failovers (clean stepdown, hard kill, frozen socket) in the normal CI matrix, with nokill -9and no hand-built infrastructure. -
Wire-level monitoring — observe every server response frame as a parsed
WireProtocolMessage, e.g. to log exactly what a server sends during a test, or to assert on protocol-level behavior your API-level test can't see. - Response manipulation — rewrite server replies before the client sees them: change topology information (that is how the failover suite works), mutate documents, or inject deliberately malformed replies to test client robustness.
It lives in morphium-core's test sources — package
de.caluga.test.morphium.testutil.proxy — so it is available to every test in this repository.
It is not (yet) published as a standalone artifact; if you want to use it outside this repo,
copy the package (it has no dependencies beyond WireProtocolMessage) or open an issue.
// Proxy in front of any wire-protocol server (MongoDB or PoppyDB)
WireProxy proxy = new WireProxy("localhost", 27017);
proxy.addObserver(new Slf4jFrameObserver()); // log every server response frame
proxy.start();
// Point the client at the proxy, not the server
MorphiumConfig cfg = new MorphiumConfig();
cfg.clusterSettings().setHostSeed("localhost:" + proxy.getListenPort());
// ... run the test ...
proxy.stop(); // severs all connections, joins every pump thread before returningWireProxy implements AutoCloseable, so try-with-resources works too. stop() guarantees
that every internal pump thread has exited before it returns — no thread leakage across tests.
The fragments above show single pieces; this is the whole thing end to end — three proxies in
front of a three-node replica set, address rewriting so the driver never escapes the proxies,
an observer logging every server reply, one write and one read flowing through, and a clean
teardown. Runs as-is from morphium-core's test scope (that's where WireProxy and
UncachedObject live):
import java.util.*;
import de.caluga.morphium.Morphium;
import de.caluga.morphium.MorphiumConfig;
import de.caluga.morphium.driver.wireprotocol.OpMsg;
import de.caluga.morphium.driver.wireprotocol.WireProtocolMessage;
import de.caluga.test.mongo.suite.data.UncachedObject;
import de.caluga.test.morphium.testutil.proxy.AddressRewriter;
import de.caluga.test.morphium.testutil.proxy.WireProxy;
public class WireProxyDemo {
public static void main(String[] args) throws Exception {
// The RS members EXACTLY as the servers report them in hello ("Map-key invariant":
// if the server says "mongo1:27017", the key must be "mongo1:27017", not an IP).
List<String> members = List.of("mongo1:27017", "mongo2:27017", "mongo3:27017");
// 1) One proxy per member, each on a random local port.
List<WireProxy> proxies = new ArrayList<>();
Map<String, String> backendToProxy = new LinkedHashMap<>();
for (String member : members) {
String host = member.substring(0, member.indexOf(':'));
int port = Integer.parseInt(member.substring(member.indexOf(':') + 1));
WireProxy proxy = new WireProxy(host, port);
proxies.add(proxy);
backendToProxy.put(member, "localhost:" + proxy.getListenPort());
}
// 2) One shared rewriter (so every hello reply, from every node, maps the full
// topology to proxy addresses) + a logging observer on every proxy.
AddressRewriter rewriter = new AddressRewriter(backendToProxy);
for (WireProxy proxy : proxies) {
proxy.setRewriter(rewriter);
proxy.addObserver((dir, msg, ctx) ->
System.out.printf("[proxy:%d] %s %s%n", ctx.listenPort(), dir, summarize(msg)));
proxy.start();
}
System.out.println("topology mapping: " + backendToProxy);
// 3) Morphium gets ONLY the proxy addresses as its seed. SSL and wire compression
// must be OFF - the proxy cannot frame-parse either (deliberate non-goal).
MorphiumConfig cfg = new MorphiumConfig();
cfg.connectionSettings().setDatabase("wireproxy_demo");
cfg.clusterSettings().getHostSeed().clear();
backendToProxy.values().forEach(cfg.clusterSettings()::addHostToSeed);
cfg.driverSettings().setDriverName("PooledDriver");
cfg.clusterSettings().setHeartbeatFrequency(1000);
cfg.driverSettings().setServerSelectionTimeout(5000);
cfg.connectionSettings().setUseSSL(false);
cfg.driverSettings().setCompressionType(MorphiumConfig.CompressionType.NONE);
// 4) Everything from here on - discovery hellos, heartbeats, the write, the read -
// shows up line by line in the observer output.
try (Morphium morphium = new Morphium(cfg)) {
morphium.store(new UncachedObject("hello through the proxy", 42));
long count = morphium.createQueryFor(UncachedObject.class).countAll();
System.out.println("read back through the proxies: " + count + " document(s)");
// Optional: watch the driver cope with a frozen node. Freeze the first proxy -
// its connections go silent (no error, no close), exactly like a paused VM.
// proxies.get(0).setFaultMode(FaultMode.freeze);
} finally {
// Severs every connection (hard RST) and joins all pump threads before returning.
for (WireProxy proxy : proxies) {
proxy.stop();
}
}
}
/** Compact one-liner per frame: hello replies show the (rewritten!) topology,
* everything else just its top-level keys. */
private static String summarize(WireProtocolMessage msg) {
if (msg instanceof OpMsg op && op.getFirstDoc() != null) {
Map<String, Object> doc = op.getFirstDoc();
if (doc.containsKey("hosts")) {
return "hello(primary=" + doc.get("primary") + ", hosts=" + doc.get("hosts") + ")";
}
return "OpMsg" + doc.keySet();
}
return msg.getClass().getSimpleName();
}
}Typical output — note that the hello lines already show proxy addresses, which is the
address rewriting doing its job; if a real backend address ever shows up here, your
backendToProxy keys don't match what the server reports:
topology mapping: {mongo1:27017=localhost:52114, mongo2:27017=localhost:52115, mongo3:27017=localhost:52116}
[proxy:52114] BACKEND_TO_CLIENT hello(primary=localhost:52114, hosts=[localhost:52114, localhost:52115, localhost:52116])
[proxy:52115] BACKEND_TO_CLIENT hello(primary=localhost:52114, hosts=[localhost:52114, localhost:52115, localhost:52116])
[proxy:52114] BACKEND_TO_CLIENT OpMsg[n, electionId, opTime, ok, ...]
[proxy:52114] BACKEND_TO_CLIENT OpMsg[cursor, ok, ...]
read back through the proxies: 1 document(s)
Faults are switched at runtime via proxy.setFaultMode(...):
FaultMode |
Existing connections | New connection attempts | Simulates |
|---|---|---|---|
passthrough |
forwarded normally | accepted | healthy network (default) |
freeze |
left open, never answered, never closed | accepted, then silence | frozen process (kill -STOP), network partition, paused VM — the failure a client cannot distinguish from a slow server |
reset |
severed with a hard RST | refused | dead process (kill -9), closed port |
close |
severed with a clean FIN | refused | this route to the node is gone (the node itself may live on — e.g. after a clean stepdown) |
Semantics worth knowing before you build a test on them:
-
freezeis one-way per connection. A connection accepted (or already open) during a freeze stays parked untilstop()— switching back topassthroughonly affects connections opened after the switch. That mirrors reality: a socket to a frozen process does not spring back to life; the client has to time out and reconnect. -
closeandresetboth refuse new connections — they differ only in how existing ones are severed (FIN vs. RST). "Nothing reachable behind this proxy right now" is the contract. -
stop()always severs with RST, regardless of the configured fault mode, so a client blocked in a read sees a definite error rather than a clean EOF that would look like an orderly shutdown.
A proxy per node is not enough for a replica set: drivers do server discovery via hello, and
the server answers with the real addresses (hosts, primary, me). After the first
hello, a driver would connect straight past your proxies.
AddressRewriter fixes that. It is a ResponseRewriter that detects hello/isMaster-shaped
replies structurally (setName + hosts present) and maps every real address to its proxy
address:
// one proxy per RS member
Map<String, String> backendToProxy = Map.of(
"mongo1:27017", "localhost:" + p1.getListenPort(),
"mongo2:27017", "localhost:" + p2.getListenPort(),
"mongo3:27017", "localhost:" + p3.getListenPort());
AddressRewriter rewriter = new AddressRewriter(backendToProxy);
p1.setRewriter(rewriter);
p2.setRewriter(rewriter);
p3.setRewriter(rewriter);The driver now lives in a consistent alternate topology consisting entirely of proxies — every
connection it ever opens, including discovery-triggered ones, flows through a fault gate.
The map keys must be the exact host:port strings the server reports (watch out for
hostname vs. IP mismatches). DriverFailoverProxyTest.assertOnlyConnectedThroughProxies shows
how to verify no traffic leaks around the proxies.
To drive the real replica set while the driver only sees proxies (e.g. trigger a genuine
replSetStepDown, poll replSetGetStatus), use ControlChannel — an auth-aware direct
connection to the real nodes, deliberately separate from the proxied data path.
proxy.addObserver((dir, msg, ctx) ->
log.info("[{}] {} -> {}", ctx.listenPort(), dir, msg.getClass().getSimpleName()));Observers are read-only by contract — they must not mutate the frame (rewriting is
ResponseRewriter's job; the interfaces are deliberately separate: fault = state,
rewrite = strategy, observe = listener). Slf4jFrameObserver is a ready-made logging
implementation. Observer exceptions never kill a proxy thread.
Today only BACKEND_TO_CLIENT frames fire: client→backend traffic is forwarded as raw,
length-prefixed bytes without parsing — deliberate pass-through fidelity, the proxy cannot
distort what it does not interpret. The CLIENT_TO_BACKEND direction exists in the enum and is
reserved for a consumer that actually needs it.
ResponseRewriter receives every parsed server reply and returns what the client should see —
including something intentionally broken:
proxy.setRewriter(reply -> {
if (reply instanceof OpMsg msg && msg.getFirstDoc() != null
&& msg.getFirstDoc().containsKey("cursor")) {
msg.getFirstDoc().put("ok", 0.0); // flip a find reply into an error
msg.getFirstDoc().put("errmsg", "injected"); // ... or corrupt it any way you like
}
return reply;
});This is the hook for robustness tests: truncated cursors, unexpected error codes, protocol violations, replies claiming a different topology than reality. Only the backend→client direction can be rewritten — requests pass through untouched by design.
- No latency injection — a frame is forwarded immediately or not at all. If you need slow-link
simulation, that would be a new
FaultMode. - No request (client→backend) rewriting or observation — see above, deliberate.
-
freezeis not reversible per connection (matches reality, but don't expect a parked connection to resume). - It is test infrastructure: no TLS termination, no config file, one backend per proxy instance.
DriverFailoverProxyTest (tag wire-failover) is the full-scale example: three proxies in
front of a real replica set, address rewriting, freeze/reset/close scenarios, stepdown via
ControlChannel, and read/write/messaging recovery assertions. It runs in the normal test
matrix against both MongoDB and PoppyDB — see the
Developer Testing Guide for how the tags fit together.