Compression at the Gateway — Response Size Optimization Strategies
In this tutorial, you'll learn about Compression. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Response compression at the gateway reduces bandwidth usage and improves client response times by compressing responses before sending them to clients.
What You'll Learn
By the end of this lesson, you will implement gzip, brotli, and zstd compression at the gateway, configure compression levels and content type policies, and tune compression for performance.
Why It Matters
Compression can reduce API response sizes by 60-80 percent, resulting in faster downloads and lower bandwidth costs for both the provider and the client.
Real-World Use
Durga Antivirus Pro compresses all JSON API responses at the gateway using brotli compression, reducing average response size from 120KB to 18KB for threat intelligence data.
Compression Flow
flowchart LR
Client-->|Accept-Encoding: gzip, br|Gateway
Gateway-->Backend[Backend Service]
Backend-->|Uncompressed Response|Gateway
Gateway-->Choose{Choose Compression}
Choose-->|br preferred|Brotli[Brotli Compress]
Choose-->|gzip|Gzip[Gzip Compress]
Choose-->|no support|Passthrough[No Compression]
Brotli-->Compressed[Compressed Response]
Gzip-->Compressed
Passthrough-->Compressed
Compressed-->Client
Compression Handler
A compression handler that supports gzip, brotli, and zstd based on client capabilities.
import gzip
import zlib
from typing import Dict, Optional, Tuple, Callable
class CompressionHandler:
SUPPORTED = {
"gzip": ("gzip", ".gz"),
"deflate": ("deflate", ".zz"),
"br": ("brotli", ".br"),
"zstd": ("zstd", ".zst"),
}
def __init__(self, default_level: int = 6,
min_size: int = 1024):
self.default_level = default_level
self.min_size = min_size
self.content_type_allowlist = {
"application/json",
"application/javascript",
"text/html",
"text/plain",
"text/css",
"text/xml",
"application/xml",
}
def negotiate(self, accept_encoding: str
) -> Optional[str]:
if not accept_encoding:
return None
encodings = [
e.strip().split(";")[0]
for e in accept_encoding.split(",")
]
for enc in encodings:
if enc in self.SUPPORTED:
return enc
return None
def should_compress(self, content_type: str,
body_size: int) -> bool:
base_type = content_type.split(";")[0].strip()
return (base_type in self.content_type_allowlist
and body_size >= self.min_size)
def compress(self, data: bytes,
encoding: str,
level: Optional[int] = None
) -> Tuple[bytes, str]:
level = level or self.default_level
if encoding == "gzip":
compressed = gzip.compress(data, level)
elif encoding == "deflate":
compressed = zlib.compress(data, level)
elif encoding == "br":
compressed = self._compress_brotli(data, level)
elif encoding == "zstd":
compressed = self._compress_zstd(data, level)
else:
return data, "identity"
return compressed, encoding
def _compress_brotli(self, data: bytes, level: int
) -> bytes:
try:
import brotli
return brotli.compress(data, quality=level)
except ImportError:
return data
def _compress_zstd(self, data: bytes, level: int
) -> bytes:
try:
import zstandard
compressor = zstandard.ZstdCompressor(level=level)
return compressor.compress(data)
except ImportError:
return data
def handle_response(self, body: bytes,
content_type: str,
accept_encoding: str
) -> Tuple[bytes, str]:
if not self.should_compress(content_type, len(body)):
return body, "identity"
encoding = self.negotiate(accept_encoding)
if not encoding:
return body, "identity"
return self.compress(body, encoding)
handler = CompressionHandler(min_size=100)
sample_body = b'{"data": "Hello World! This is a test response."}'
original_size = len(sample_body)
compressed, encoding = handler.handle_response(
sample_body, "application/json",
"gzip, br"
)
print(f"Original: {original_size}B, Compressed ({encoding}): "
f"{len(compressed)}B, Ratio: {len(compressed)/original_size:.2%}")
Compression Level Tuning
Balance compression ratio against CPU cost by choosing the right compression level.
import time
import gzip
from typing import Dict, Tuple
class CompressionBenchmark:
def __init__(self):
self.results: Dict[int, Tuple[float, int]] = {}
def benchmark_level(self, data: bytes, level: int
) -> Tuple[float, int]:
start = time.time()
compressed = gzip.compress(data, level)
duration = time.time() - start
self.results[level] = (duration, len(compressed))
return duration, len(compressed)
def recommend_level(self) -> int:
best_ratio = float("inf")
best_level = 6
for level, (duration, size) in self.results.items():
score = duration * size
if score < best_ratio:
best_ratio = score
best_level = level
return best_level
def print_report(self, original_size: int):
print(f"{'Level':>6} | {'Time (ms)':>10} | "
f"{'Size':>8} | {'Ratio':>6}")
for level, (duration, size) in sorted(self.results.items()):
ratio = size / original_size * 100
print(f"{level:6d} | {duration*1000:10.2f} | "
f"{size:8d} | {ratio:5.1f}%")
benchmark = CompressionBenchmark()
data = b"x" * 100000 + b'{"data": "test" * 10000}'
for level in [1, 3, 6, 9]:
benchmark.benchmark_level(data, level)
benchmark.print_report(len(data))
print(f"Recommended level: {benchmark.recommend_level()}")
Per-Route Compression Policy
Configure different compression strategies for different API routes.
from typing import Dict, Optional, Tuple
import re
class RouteCompressionPolicy:
def __init__(self):
self.policies: Dict[str, Dict] = {}
def add_policy(self, path_pattern: str,
enabled: bool = True,
min_size: Optional[int] = None,
encoding: Optional[str] = None,
level: Optional[int] = None):
self.policies[path_pattern] = {
"enabled": enabled,
"min_size": min_size,
"encoding": encoding,
"level": level,
}
def get_policy(self, path: str) -> Dict:
for pattern, policy in self.policies.items():
if re.search(pattern, path):
return policy
return {"enabled": True}
def should_compress(self, path: str, size: int
) -> bool:
policy = self.get_policy(path)
if not policy.get("enabled", True):
return False
min_size = policy.get("min_size", 1024)
if size < min_size:
return False
return True
policy = RouteCompressionPolicy()
policy.add_policy(r"^/api/stream", enabled=False)
policy.add_policy(r"^/api/reports", min_size=5120, level=9)
policy.add_policy(r"^/api/health", enabled=True, min_size=0)
for path in ["/api/health", "/api/stream/video", "/api/reports/daily"]:
should = policy.should_compress(path, 2000)
print(f"{path}: compress={should}")
Common Mistakes
Mistake 1: Compressing Already Compressed Data
Compressing data that is already compressed (images, video) wastes CPU and may increase size. Check content type.
Mistake 2: Ignoring Content Negotiation
Always respect the Accept-Encoding header. Compressing when the client does not support it wastes resources.
Mistake 3: Using Maximum Compression Level
Level 9 compression saves a few percent more than level 6 but uses significantly more CPU. Level 6 is generally optimal.
Mistake 4: Compressing Small Payloads
Compression adds overhead. Payloads under 1KB may become larger after compression due to headers and dictionary overhead.
Mistake 5: Not Setting Content-Encoding Header
Without the Content-Encoding header, clients cannot decompress the response. Always set it correctly.
Practice Questions
- What is the difference between gzip and brotli compression?
- How does the Accept-Encoding header determine which compression to use?
- Why is compression level 6 recommended over level 9 for APIs?
- What content types benefit most from compression?
- How do you handle compression for streaming responses?
Challenge
Build a compression middleware for the gateway that negotiates the best compression algorithm from the Accept-Encoding header, compresses JSON and text responses over 1KB, supports gzip and brotli, and sets the correct Content-Encoding header.
FAQ
Mini Project
Build a compression plugin for the gateway that supports gzip, brotli, and zstd, negotiates via Accept-Encoding, applies configurable compression levels per route, sets correct Content-Encoding and Vary headers, and skips compression for small payloads and binary content types.
What's Next
Learn about Caching Deep for response Caching strategies, or explore SSL Termination Deep for secure connection management.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro