mirror of
https://github.com/fastapi/fastapi.git
synced 2026-10-09 11:41:39 -04:00
37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
import threading
|
|
from contextlib import contextmanager
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from socketserver import TCPServer
|
|
|
|
|
|
class _CollectorServer(ThreadingHTTPServer):
|
|
def server_bind(self):
|
|
TCPServer.server_bind(self)
|
|
self.server_name = str(self.server_address[0])
|
|
self.server_port = self.server_address[1]
|
|
|
|
|
|
@contextmanager
|
|
def otlp_collector():
|
|
received = []
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
body = self.rfile.read(int(self.headers["Content-Length"]))
|
|
received.append((self.path, body, self.headers))
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/x-protobuf")
|
|
self.end_headers()
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
with _CollectorServer(("127.0.0.1", 0), Handler) as server:
|
|
worker = threading.Thread(target=server.serve_forever, daemon=True)
|
|
worker.start()
|
|
try:
|
|
yield f"http://127.0.0.1:{server.server_port}", received
|
|
finally:
|
|
server.shutdown()
|
|
worker.join()
|