JMS, FTP, LDAP, TCP, and Other Protocol Test Patterns: Guided Hands-On Workflow

1. Safety envelope

2. Create the dependency-free TCP fixture

Save as fixtures/tcp_line_fixture.py:

from socketserver import ThreadingTCPServer, StreamRequestHandler
from pathlib import Path
import argparse
import itertools
import json
import threading
import time

FIXTURE_VERSION = "prompt17-tcp-line-fixture-v1"
connection_seq = itertools.count(1)
event_lock = threading.Lock()
event_log = None
active_connections = set()

def now_ms():
    return int(time.time() * 1000)

def log_event(event):
    if event_log is None:
        return
    with event_lock:
        with event_log.open("a", encoding="utf-8") as handle:
            handle.write(json.dumps(event, sort_keys=True) + "\n")

class Handler(StreamRequestHandler):
    def setup(self):
        super().setup()
        self.conn_id = f"C{next(connection_seq):05d}"
        with event_lock:
            active_connections.add(self.conn_id)
        log_event({
            "ts_ms": now_ms(),
            "event": "connect",
            "conn_id": self.conn_id,
            "active_connections": len(active_connections),
        })

    def handle(self):
        while True:
            raw = self.rfile.readline(8192)
            if not raw:
                break

            started = now_ms()
            try:
                line = raw.decode("utf-8").rstrip("\r\n")
            except UnicodeDecodeError:
                response = f"ERR|CONN={self.conn_id}|CODE=BAD_UTF8\n"
                self.wfile.write(response.encode("utf-8"))
                self.wfile.flush()
                log_event({
                    "ts_ms": now_ms(), "event": "request", "conn_id": self.conn_id,
                    "command": "BAD_UTF8", "status": "ERR",
                    "service_wall_ms": now_ms() - started
                })
                continue

            parts = line.split("|", 3)
            command = parts[0].upper() if parts else ""
            req_id = parts[1] if len(parts) > 1 else ""

            if command == "PING":
                payload = parts[2] if len(parts) > 2 else ""
                response = (
                    f"OK|CONN={self.conn_id}|REQ={req_id}|"
                    f"CMD=PING|PAYLOAD={payload}\n"
                )
                status = "OK"
                close_after = False
            elif command == "ECHO":
                payload = parts[2] if len(parts) > 2 else ""
                response = (
                    f"OK|CONN={self.conn_id}|REQ={req_id}|"
                    f"CMD=ECHO|PAYLOAD={payload}\n"
                )
                status = "OK"
                close_after = False
            elif command == "DELAY":
                try:
                    delay_ms = int(parts[2]) if len(parts) > 2 else 0
                except ValueError:
                    delay_ms = -1
                if 0 <= delay_ms <= 100:
                    time.sleep(delay_ms / 1000.0)
                    response = (
                        f"OK|CONN={self.conn_id}|REQ={req_id}|"
                        f"CMD=DELAY|DELAY_MS={delay_ms}\n"
                    )
                    status = "OK"
                else:
                    response = (
                        f"ERR|CONN={self.conn_id}|REQ={req_id}|"
                        f"CODE=BAD_DELAY\n"
                    )
                    status = "ERR"
                close_after = False
            elif command == "CLOSE":
                response = (
                    f"BYE|CONN={self.conn_id}|REQ={req_id}|CMD=CLOSE\n"
                )
                status = "BYE"
                close_after = True
            else:
                response = (
                    f"ERR|CONN={self.conn_id}|REQ={req_id}|"
                    f"CODE=UNKNOWN_COMMAND\n"
                )
                status = "ERR"
                close_after = False

            self.wfile.write(response.encode("utf-8"))
            self.wfile.flush()
            log_event({
                "ts_ms": now_ms(),
                "event": "request",
                "conn_id": self.conn_id,
                "command": command,
                "req_id": req_id,
                "status": status,
                "service_wall_ms": now_ms() - started,
            })

            if close_after:
                break

    def finish(self):
        try:
            with event_lock:
                active_connections.discard(self.conn_id)
                active = len(active_connections)
            log_event({
                "ts_ms": now_ms(),
                "event": "disconnect",
                "conn_id": self.conn_id,
                "active_connections": active,
            })
        finally:
            super().finish()

class Server(ThreadingTCPServer):
    allow_reuse_address = True
    daemon_threads = True

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=9100)
    parser.add_argument("--log", default="results/tcp-events.jsonl")
    args = parser.parse_args()

    global event_log
    event_log = Path(args.log).resolve()
    event_log.parent.mkdir(parents=True, exist_ok=True)
    event_log.write_text("", encoding="utf-8")

    with Server((args.host, args.port), Handler) as server:
        print(f"fixture_version={FIXTURE_VERSION}")
        print(f"listen={args.host}:{args.port}")
        print(f"event_log={event_log}")
        server.serve_forever()

if __name__ == "__main__":
    main()

Start it:

PowerShell:

New-Item -ItemType Directory -Force .\results | Out-Null
python .\fixtures\tcp_line_fixture.py `
  --host 127.0.0.1 `
  --port 9100 `
  --log .\results\tcp-events.jsonl

Bash:

mkdir -p results
python fixtures/tcp_line_fixture.py   --host 127.0.0.1   --port 9100   --log results/tcp-events.jsonl

The server assigns each accepted socket a synthetic CONN ID and writes connect/request/disconnect events. It does not record passwords or sensitive data because none exist.

3. Prove the line protocol before JMeter

Use this short Python one-shot client:

import socket

with socket.create_connection(("127.0.0.1", 9100), timeout=1) as s:
    f = s.makefile("rwb")
    f.write(b"PING|manual-1|hello\n")
    f.flush()
    print(f.readline().decode().strip())

Expected shape:

OK|CONN=C00001|REQ=manual-1|CMD=PING|PAYLOAD=hello

This proves target/framing independently of JMeter.

4. Configure TCP Sampler Config

EOL byte configures response reading. The request still needs a line terminator, so use JMeter's __char(10) function in Text to Send.

5. Create per-sample request IDs

Use text that combines the JMeter thread number and current time:

PING|T${__threadNum}-${__time()}|hello-${__threadNum}${__char(10)}

The request ID is diagnostic correlation state, not security state. The server echoes it so JTL response and server event can be matched.

6. First TCP Request and connection extraction

Add TCP Sampler — TCP Ping A using the config. Add Response Assertion containing:

• OK|CONN=
• |CMD=PING|
• PAYLOAD=hello-${__threadNum}

Add Regular Expression Extractor:

Reference Name: CONN_A
Regular Expression: CONN=([^|\r\n]+)
Template: $1$
Match No.: 1
Default: __NO_CONN__

7. Second sampler proves same-thread reuse

Add TCP Sampler — TCP Echo B:

ECHO|T${__threadNum}-${__time()}|second-${__threadNum}${__char(10)}

Add Response Assertion containing CONN=${CONN_A} and CMD=ECHO. Because the exact same host/port and same JMeter thread are used with reuse enabled, the connection ID should remain the same.

8. Baseline tree

Test Plan
├── TCP Sampler Config
│   127.0.0.1:9100
│   TCPClientImpl
│   Re-use=true
│   EOL=10
└── Thread Group — 2 threads × 5 loops
    ├── TCP Ping A
    │   └── RegEx Extractor -> CONN_A
    └── TCP Echo B
        └── assert response contains CONN=${CONN_A}

Configured workload = 20 request/reply samples (2 threads ×5 loops ×2 samplers).

9. One-thread authoring check

Before concurrency, use 1 thread ×1 loop in GUI with View Results Tree temporarily enabled. Confirm two response lines, same CONN ID, correct payload, no timeout, then disable the listener.

10. CLI baseline

PowerShell:

jmeter.bat -n `
  -t plans\tcp-persistent.jmx `
  -l results\persistent\results.jtl `
  -j results\persistent\jmeter.log `
  -e -o results\persistent\report

Preserve JTL and matching jmeter.log; record generator CPU/memory/socket count during the tiny run.

11. Analyze target connection evidence

Save tools/analyze_tcp_events.py:

import json
import sys
from collections import Counter, defaultdict
from pathlib import Path

path = Path(sys.argv[1] if len(sys.argv) > 1 else "results/tcp-events.jsonl")
events = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
if not events:
    raise SystemExit("No TCP events found")

connects = [e for e in events if e["event"] == "connect"]
requests = [e for e in events if e["event"] == "request"]
disconnects = [e for e in events if e["event"] == "disconnect"]

print(f"connections_opened={len(connects)}")
print(f"requests={len(requests)}")
print(f"disconnects={len(disconnects)}")
print(f"commands={dict(Counter(e.get('command','') for e in requests))}")
print(f"statuses={dict(Counter(e.get('status','') for e in requests))}")

by_conn = defaultdict(list)
for e in requests:
    by_conn[e["conn_id"]].append(e)

print("requests_per_connection:")
for conn, items in sorted(by_conn.items()):
    print(f"  {conn}: {len(items)} requests")

max_active = max(
    [e.get("active_connections", 0) for e in events if "active_connections" in e] or [0]
)
print(f"max_active_connections={max_active}")

req_ids = [e.get("req_id","") for e in requests if e.get("req_id")]
print(f"unique_request_ids={len(set(req_ids))}")
print(f"duplicate_request_ids={len(req_ids)-len(set(req_ids))}")

python tools/analyze_tcp_events.py results/tcp-events.jsonl

With 2 persistent JMeter threads, expect approximately two JMeter TCP connections and multiple requests per connection. Manual preflight connections are separate and should be archived/cleared or labeled before the measured run.

12. Per-sample connection comparison

Duplicate the plan, keep 2×5 and payloads identical, then uncheck Re-use connection. Run separately as results/per-sample/....

Prediction: server logs approximately one TCP connection per sampler (about 20 for the measured JMeter workload), versus approximately two persistent JMeter connections in the reused profile. The response/business work is the same; only socket lifecycle changed.

13. Request/reply timing exercise

Use a third small profile with 1 thread ×5 loops:

DELAY|T${__threadNum}-${__time()}|25${__char(10)}

The fixture sleeps 25 ms before replying. JMeter sampler elapsed should be at least roughly the server delay plus local/client overhead. This is a true request/reply boundary because the sampler waits for the line reply.

14. Cleanup and connection lifecycle

JMeter closes TCP sockets at test end. The fixture logs disconnect events. Stop the fixture only after the run and verify:

• no active fixture connections;
• connection count matches the selected reuse model approximately;
• all configured request IDs received responses;
• no unexpected ERR status.

15. Guided architecture exercise — JMS request/reply

Do not install a broker just to complete the chapter. Design the equivalent local architecture:

JMeter JMS Point-to-Point
        |
        | request message + correlation ID
        v
temporary/request queue
        |
        v
local synthetic service consumer
        |
        | reply message carrying same correlation ID
        v
temporary/reply queue
        |
        v
JMeter waits in Request Response mode

Required design state:

• provider-specific JMS client/JNDI JARs pinned and copied to JMeter lib;
• localhost-only broker/provider URL;
• dedicated run-scoped request/reply queues or selectors;
• fake credentials only;
• request-reply mode for end-to-reply timing;
• queue depth before/after and cleanup evidence;
• unique correlation/client IDs per thread where the provider requires uniqueness.

A request-only alternative is valid when the metric is producer/send performance—but it must never be reported as consumer end-to-end processing latency.

16. FTP and LDAP architecture notes

FTP: use a local disposable server root and fake account; decide GET versus PUT, binary versus ASCII, local-file I/O, hash/size assertion, and explicit removal of uploaded synthetic files. Keep in mind FTP sampler latency is login time.

LDAP: use a disposable local directory; for realistic sessions, plan thread bind → search/compare operations → thread unbind. Keep mandatory exercises search-only unless a dedicated synthetic directory has explicit cleanup.

17. Challenge

You need to measure a service that receives JMS requests, performs work, then places a reply on another queue. Which JMeter timing model should be the primary metric?

Use JMS Point-to-Point Request Response with a dedicated/correlated reply path. Request Only measures producer send performance, not service request-to-reply latency.

Knowledge check

Official references and version notes

• JMeter Component Reference — TCP Sampler — built-in TCP clients, EOL/binary framing, per-thread connection reuse, timeouts, and socket lifecycle.
• JMeter Component Reference — FTP Request — upload/download behavior, login latency, binary/ASCII modes, and credential storage.
• JMeter Component Reference — JMS Publisher — provider/JNDI dependencies, destinations, persistence, message sources, and provider JAR requirements.
• JMeter Component Reference — JMS Subscriber — queue/topic clients, durable IDs, selectors, receive/listener strategies, and connection lifecycle.
• JMeter Component Reference — JMS Point-to-Point — request-only, request-reply, read, browse, and clear semantics.
• JMeter Component Reference — LDAP Request — Add/Modify/Delete/Search operations and cleanup differences.
• JMeter Component Reference — LDAP Extended Request — bind/session-oriented LDAP operations.
• JMeter Functions — __char — inserts explicit Unicode characters such as LF for TCP text framing.
• JMeter Getting Started — Java requirements and GUI authoring versus CLI execution.
• JMeter Best Practices — non-GUI execution and generator validity.
• Apache JMeter downloads — current stable release and Java requirement.