Web Server

def server(sock, site, log=None, environ=None, max_size=None, 
          max_http_version=None, protocol=wsgi.HttpProtocol, 
          server_event=None, minimum_chunk_size=None, 
          log_x_forwarded_for=True, custom_pool=None, 
          keepalive=True, log_output=True, log_format=None, 
          url_length_limit=8192, debug=True, socket_timeout=None, 
          capitalize_response_headers=True):
    """
    Start a WSGI server handling requests on the given socket.
    
    Parameters:
    - sock: listening socket to accept connections from
    - site: WSGI application callable
    - log: file-like object for access logging (default: sys.stdout)
    - environ: dict of additional WSGI environ variables
    - max_size: int, maximum number of concurrent connections
    - max_http_version: str, maximum HTTP version to support
    - protocol: protocol class for handling HTTP requests
    - server_event: Event object signaled when server starts
    - minimum_chunk_size: int, minimum chunk size for chunked encoding
    - log_x_forwarded_for: bool, whether to log X-Forwarded-For header
    - custom_pool: GreenPool instance for connection handling
    - keepalive: bool, whether to support HTTP keep-alive
    - log_output: bool, whether to enable access logging
    - log_format: str, custom log format string
    - url_length_limit: int, maximum URL length in bytes
    - debug: bool, whether to include debug information in errors
    - socket_timeout: int or None, socket timeout in seconds (default: None)
    - capitalize_response_headers: bool, whether to capitalize response headers
    
    Returns:
    None (runs until interrupted)
    """

def format_date_time(timestamp):
    """
    Format a unix timestamp to HTTP standard date/time string.
    
    Parameters:
    - timestamp: float, unix timestamp
    
    Returns:
    str: formatted date string for HTTP headers
    """

class WebSocketWSGI:
    """
    Wraps a websocket handler function in a WSGI application.
    Enables WebSocket support within WSGI frameworks.
    """
    
    def __init__(self, handler):
        """
        Create a WebSocket WSGI wrapper.
        
        Parameters:
        - handler: callable(ws) that handles WebSocket connections
        """
    
    def __call__(self, environ, start_response):
        """
        WSGI application interface.
        
        Parameters:
        - environ: WSGI environment dict
        - start_response: WSGI start_response callable
        
        Returns:
        WSGI response iterable
        """

class WebSocket:
    """
    WebSocket connection object for sending and receiving messages.
    """
    
    def send(self, message):
        """
        Send a message to the WebSocket client.
        
        Parameters:
        - message: str or bytes, message to send
        
        Returns:
        None
        
        Raises:
        ConnectionClosed: if WebSocket connection is closed
        """
    
    def receive(self):
        """
        Receive a message from the WebSocket client.
        
        Returns:
        str or bytes: received message
        
        Raises:
        ConnectionClosed: if WebSocket connection is closed
        """
    
    def close(self):
        """
        Close the WebSocket connection.
        
        Returns:
        None
        """
    
    @property
    def closed(self):
        """
        Check if the WebSocket connection is closed.
        
        Returns:
        bool: True if closed, False if open
        """

import eventlet
import eventlet.wsgi

def hello_world_app(environ, start_response):
    """Simple WSGI application"""
    status = '200 OK'
    headers = [('Content-Type', 'text/plain')]
    start_response(status, headers)
    return [b'Hello, World!']

def run_wsgi_server():
    """Run a basic WSGI server"""
    # Create listening socket
    sock = eventlet.listen(('localhost', 8080))
    print("WSGI server listening on localhost:8080")
    
    # Start WSGI server
    eventlet.wsgi.server(sock, hello_world_app)

if __name__ == "__main__":
    run_wsgi_server()

import eventlet
import eventlet.wsgi
from flask import Flask

app = Flask(__name__)

@app.route('/')
def hello():
    return 'Hello from Flask on Eventlet!'

@app.route('/users/<int:user_id>')
def get_user(user_id):
    return f'User ID: {user_id}'

@app.route('/api/data')
def api_data():
    # Simulate some async work
    eventlet.sleep(0.1)
    return {'message': 'Data from API', 'status': 'success'}

def run_flask_server():
    """Run Flask application with Eventlet WSGI server"""
    sock = eventlet.listen(('localhost', 5000))
    print("Flask server with Eventlet listening on localhost:5000")
    
    eventlet.wsgi.server(
        sock, 
        app,
        log_output=True,
        max_size=1000  # Max 1000 concurrent connections
    )

if __name__ == "__main__":
    # Enable monkey patching for Flask compatibility
    eventlet.monkey_patch()
    run_flask_server()

import eventlet
import eventlet.wsgi
import eventlet.websocket

# Store connected clients
clients = set()

def websocket_handler(ws):
    """Handle WebSocket connections"""
    clients.add(ws)
    print(f"Client connected. Total clients: {len(clients)}")
    
    try:
        while not ws.closed:
            message = ws.receive()
            if message is None:
                break
            
            print(f"Received: {message}")
            
            # Broadcast message to all connected clients
            disconnected = set()
            for client in clients:
                try:
                    client.send(f"Broadcast: {message}")
                except:
                    disconnected.add(client)
            
            # Remove disconnected clients
            clients -= disconnected
            
    except eventlet.websocket.ConnectionClosed:
        pass
    finally:
        clients.discard(ws)
        print(f"Client disconnected. Total clients: {len(clients)}")

def websocket_app(environ, start_response):
    """WSGI app that handles WebSocket upgrades"""
    if environ.get('HTTP_UPGRADE', '').lower() == 'websocket':
        # Handle WebSocket upgrade
        return eventlet.websocket.WebSocketWSGI(websocket_handler)(environ, start_response)
    else:
        # Handle regular HTTP requests
        status = '200 OK'
        headers = [('Content-Type', 'text/html')]
        start_response(status, headers)
        
        return [b"""
<!DOCTYPE html>
<html>
<head><title>WebSocket Chat</title></head>
<body>
    <div id="messages"></div>
    <input type="text" id="messageInput" placeholder="Type a message...">
    <button onclick="sendMessage()">Send</button>
    
    <script>
        const ws = new WebSocket('ws://localhost:8080');
        const messages = document.getElementById('messages');
        
        ws.onmessage = function(event) {
            const div = document.createElement('div');
            div.textContent = event.data;
            messages.appendChild(div);
        };
        
        function sendMessage() {
            const input = document.getElementById('messageInput');
            ws.send(input.value);
            input.value = '';
        }
        
        document.getElementById('messageInput').addEventListener('keypress', function(e) {
            if (e.key === 'Enter') {
                sendMessage();
            }
        });
    </script>
</body>
</html>
        """]

def run_websocket_server():
    """Run WebSocket chat server"""
    sock = eventlet.listen(('localhost', 8080))
    print("WebSocket chat server listening on localhost:8080")
    
    eventlet.wsgi.server(sock, websocket_app, debug=True)

if __name__ == "__main__":
    eventlet.monkey_patch()
    run_websocket_server()

import eventlet
import eventlet.wsgi
import logging
import sys

def create_wsgi_app():
    """Create a sample WSGI application"""
    def app(environ, start_response):
        path = environ.get('PATH_INFO', '/')
        
        if path == '/':
            status = '200 OK'
            headers = [('Content-Type', 'text/html')]
            start_response(status, headers)
            return [b'<h1>Production WSGI Server</h1>']
        
        elif path == '/health':
            status = '200 OK' 
            headers = [('Content-Type', 'application/json')]
            start_response(status, headers)
            return [b'{"status": "healthy"}']
        
        else:
            status = '404 Not Found'
            headers = [('Content-Type', 'text/plain')]
            start_response(status, headers)
            return [b'Page not found']
    
    return app

def run_production_server():
    """Run production WSGI server with full configuration"""
    # Set up logging
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    
    # Create application
    app = create_wsgi_app()
    
    # Create listening socket
    sock = eventlet.listen(('0.0.0.0', 8000))
    print("Production WSGI server listening on 0.0.0.0:8000")
    
    # Configure custom environment variables
    custom_environ = {
        'SERVER_NAME': 'eventlet-server',
        'SERVER_SOFTWARE': 'Eventlet/1.0'
    }
    
    # Start server with production configuration
    eventlet.wsgi.server(
        sock,
        app,
        log=sys.stdout,  # Access log output
        environ=custom_environ,
        max_size=2000,  # Max concurrent connections
        keepalive=True,  # Enable HTTP keep-alive
        log_x_forwarded_for=True,  # Log X-Forwarded-For header
        url_length_limit=16384,  # 16KB URL limit
        socket_timeout=30,  # 30 second socket timeout
        debug=False,  # Disable debug mode for production
        capitalize_response_headers=True
    )

if __name__ == "__main__":
    eventlet.monkey_patch()
    run_production_server()

import eventlet
import eventlet.wsgi
import logging
import time
from io import StringIO

class WSGILogger:
    """Custom WSGI access logger"""
    
    def __init__(self, logger):
        self.logger = logger
    
    def write(self, msg):
        """Write log message"""
        if msg.strip():  # Ignore empty messages
            self.logger.info(msg.strip())
    
    def flush(self):
        """Flush any buffered output"""
        pass

def timed_wsgi_app(environ, start_response):
    """WSGI app that tracks request timing"""
    start_time = time.time()
    
    # Process request
    path = environ.get('PATH_INFO', '/')
    
    if path.startswith('/slow'):
        # Simulate slow endpoint
        eventlet.sleep(2.0)
        response = b'Slow response completed'
    else:
        response = b'Fast response'
    
    # Calculate processing time
    processing_time = time.time() - start_time
    
    status = '200 OK'
    headers = [
        ('Content-Type', 'text/plain'),
        ('X-Processing-Time', f'{processing_time:.3f}s')
    ]
    start_response(status, headers)
    
    return [response]

def run_logged_server():
    """Run server with custom logging"""
    # Set up logger
    logger = logging.getLogger('wsgi_access')
    logger.setLevel(logging.INFO)
    
    handler = logging.StreamHandler()
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(message)s'
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    
    # Create custom log writer
    log_writer = WSGILogger(logger)
    
    # Start server
    sock = eventlet.listen(('localhost', 8080))
    print("Logged WSGI server listening on localhost:8080")
    
    eventlet.wsgi.server(
        sock,
        timed_wsgi_app,
        log=log_writer,
        log_format='%(client_ip)s - "%(request_line)s" %(status_code)s %(body_length)s %(wall_seconds).6f',
        log_output=True
    )

if __name__ == "__main__":
    eventlet.monkey_patch()
    run_logged_server()

import eventlet
import eventlet.wsgi
import ssl

def https_app(environ, start_response):
    """WSGI app for HTTPS server"""
    status = '200 OK'
    headers = [
        ('Content-Type', 'text/html'),
        ('Strict-Transport-Security', 'max-age=31536000; includeSubDomains')
    ]
    start_response(status, headers)
    
    return [b'''
    <html>
    <head><title>Secure HTTPS Server</title></head>
    <body>
        <h1>Secure Connection</h1>
        <p>This page is served over HTTPS using Eventlet.</p>
    </body>
    </html>
    ''']

def run_https_server():
    """Run HTTPS WSGI server"""
    # Create socket and wrap with SSL
    sock = eventlet.listen(('localhost', 8443))
    
    # Wrap socket with SSL
    ssl_sock = eventlet.wrap_ssl(
        sock,
        certfile='server.crt',  # SSL certificate file
        keyfile='server.key',   # SSL private key file
        server_side=True,
        ssl_version=ssl.PROTOCOL_TLS,
        ciphers='ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM:DHE+CHACHA20:!aNULL:!MD5:!DSS'
    )
    
    print("HTTPS WSGI server listening on https://localhost:8443")
    
    eventlet.wsgi.server(
        ssl_sock,
        https_app,
        log_output=True,
        debug=False
    )

if __name__ == "__main__":
    eventlet.monkey_patch()
    run_https_server()

# High-concurrency configuration
eventlet.wsgi.server(
    sock,
    app,
    max_size=5000,          # Support 5000 concurrent connections
    socket_timeout=120,     # 2 minute timeout
    keepalive=True,         # Enable keep-alive for efficiency
    minimum_chunk_size=8192 # Larger chunks for better performance
)

# Memory-constrained configuration  
eventlet.wsgi.server(
    sock,
    app,
    max_size=500,           # Limit concurrent connections
    socket_timeout=30,      # Shorter timeout
    url_length_limit=4096   # Smaller URL limit
)

import eventlet.wsgi

class CustomProtocol(eventlet.wsgi.HttpProtocol):
    """Custom HTTP protocol with additional features"""
    
    def handle_one_request(self):
        """Override to add custom request handling"""
        # Add custom headers or logging
        result = super().handle_one_request()
        return result

# Use custom protocol
eventlet.wsgi.server(
    sock,
    app,
    protocol=CustomProtocol
)