diff --git a/.github/ISSUE_TEMPLATE/bug_report.md b/.github/ISSUE_TEMPLATE/bug_report.md new file mode 100644 index 0000000..5ace4d7 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/bug_report.md @@ -0,0 +1,42 @@ +--- +name: Bug report +about: Create a report to help us improve +title: '[BUG] ' +labels: bug +assignees: '' + +--- + +**Describe the bug** +A clear and concise description of what the bug is. + +**To Reproduce** +Steps to reproduce the behavior: +1. Configure server with '...' +2. Send message to '....' +3. Observe error '....' + +**Expected behavior** +A clear and concise description of what you expected to happen. + +**Environment:** + - OS: [e.g. Ubuntu 22.04, macOS 14, Windows 11] + - AMQP-Go Version: [e.g. v0.1.0] + - Go Version: [e.g. 1.25.1] + - Storage Backend: [e.g. memory, badger] + +**Configuration:** +```json +{ + // Paste your configuration here (remove sensitive data) +} +``` + +**Logs:** +``` +Paste relevant log output here +``` + +**Additional context** +Add any other context about the problem here. + diff --git a/.github/ISSUE_TEMPLATE/feature_request.md b/.github/ISSUE_TEMPLATE/feature_request.md new file mode 100644 index 0000000..3a3c83a --- /dev/null +++ b/.github/ISSUE_TEMPLATE/feature_request.md @@ -0,0 +1,24 @@ +--- +name: Feature request +about: Suggest an idea for this project +title: '[FEATURE] ' +labels: enhancement +assignees: '' + +--- + +**Is your feature request related to a problem? Please describe.** +A clear and concise description of what the problem is. Ex. I'm always frustrated when [...] + +**Describe the solution you'd like** +A clear and concise description of what you want to happen. + +**Describe alternatives you've considered** +A clear and concise description of any alternative solutions or features you've considered. + +**Additional context** +Add any other context or screenshots about the feature request here. + +**AMQP Compliance** +If this feature relates to AMQP 0.9.1 compliance, please reference the relevant section of the specification. + diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..b3b08de --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,29 @@ +version: 2 +updates: + # Enable version updates for Go modules + - package-ecosystem: "gomod" + directory: "/src/amqp-go" + schedule: + interval: "weekly" + day: "monday" + open-pull-requests-limit: 10 + labels: + - "dependencies" + - "go" + commit-message: + prefix: "chore(deps)" + include: "scope" + + # Enable version updates for GitHub Actions + - package-ecosystem: "github-actions" + directory: "/" + schedule: + interval: "weekly" + day: "monday" + open-pull-requests-limit: 5 + labels: + - "dependencies" + - "github-actions" + commit-message: + prefix: "chore(ci)" + diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md new file mode 100644 index 0000000..542b42b --- /dev/null +++ b/.github/pull_request_template.md @@ -0,0 +1,55 @@ +## Description + + +## Type of Change + +- [ ] Bug fix (non-breaking change which fixes an issue) +- [ ] New feature (non-breaking change which adds functionality) +- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected) +- [ ] Documentation update +- [ ] Performance improvement +- [ ] Code refactoring + +## Related Issues + +Closes # + +## Changes Made + +- +- +- + +## Testing + +- [ ] All existing tests pass +- [ ] Added new tests for changes +- [ ] Tested manually with the following scenarios: + - + - + +## Performance Impact + +- [ ] No performance impact +- [ ] Performance improvement (provide benchmark results) +- [ ] Performance regression (explain why acceptable) + +## Documentation + +- [ ] Updated README.md +- [ ] Updated code comments +- [ ] Updated examples +- [ ] No documentation needed + +## Checklist +- [ ] My code follows the project's style guidelines +- [ ] I have performed a self-review of my own code +- [ ] I have commented my code, particularly in hard-to-understand areas +- [ ] My changes generate no new warnings or errors +- [ ] I have added tests that prove my fix is effective or that my feature works +- [ ] New and existing unit tests pass locally with my changes +- [ ] Any dependent changes have been merged and published + +## Screenshots (if applicable) + + diff --git a/.github/workflows/README.md b/.github/workflows/README.md new file mode 100644 index 0000000..a61ce87 --- /dev/null +++ b/.github/workflows/README.md @@ -0,0 +1,196 @@ +# GitHub Actions Workflows + +This directory contains automated workflows for the AMQP-Go project. + +## Workflows + +### 1. Release Build (`release.yml`) + +**Triggers:** +- When a new release is created on GitHub +- Manual dispatch with version input + +**What it does:** +- Builds binaries for all supported platforms: + - macOS: arm64, amd64 + - Linux: amd64, arm64, 386 +- Generates SHA256 checksums for each binary +- Uploads all binaries to the GitHub release + +**Platforms:** +- macOS builds run on: `macos-latest` +- Linux cross-compilation runs on: `ubuntu-latest` + +**Usage:** +```bash +# Create a release +git tag -a v0.1.0 -m "Release v0.1.0" +git push origin v0.1.0 + +# Then create release on GitHub +# Workflow will automatically trigger +``` + +**Manual trigger:** +- Go to Actions → Release Build → Run workflow +- Enter version tag (e.g., v0.1.0-dev) + +### 2. Build and Test (`build.yml`) + +**Triggers:** +- Push to `main` or `develop` branches +- Pull requests to `main` or `develop` + +**What it does:** +- Runs full test suite with race detection +- Generates code coverage report +- Uploads coverage to Codecov +- Runs golangci-lint for code quality +- Builds the binary to verify compilation +- Caches Go modules for faster builds + +**Jobs:** +1. **Test**: Run tests with coverage +2. **Lint**: Static code analysis +3. **Build**: Verify binary builds + +### 3. CodeQL Analysis (`codeql.yml`) + +**Triggers:** +- Push to `main` branch +- Pull requests to `main` +- Weekly schedule (Monday) + +**What it does:** +- Performs security and quality analysis +- Identifies potential vulnerabilities +- Checks for common coding mistakes +- Reports findings in Security tab + +### 4. Dependabot (`dependabot.yml`) + +**Schedule:** Weekly on Monday + +**What it does:** +- Checks for Go module updates +- Checks for GitHub Actions updates +- Creates PRs for outdated dependencies +- Keeps dependencies secure and up-to-date + +## Secrets Required + +### For Codecov (Optional) + +- `CODECOV_TOKEN`: Token from codecov.io (optional but recommended) + +## Workflow Badges + +Add these to your README.md: + +```markdown +[![Build Status](https://github.com/YOUR-ORG/strangeq/workflows/Build%20and%20Test/badge.svg)](https://github.com/YOUR-ORG/strangeq/actions) +[![CodeQL](https://github.com/YOUR-ORG/strangeq/workflows/CodeQL/badge.svg)](https://github.com/YOUR-ORG/strangeq/actions) +[![codecov](https://codecov.io/gh/YOUR-ORG/strangeq/branch/main/graph/badge.svg)](https://codecov.io/gh/YOUR-ORG/strangeq) +``` + +## Local Testing + +### Test the build locally + +```bash +# Install act (GitHub Actions local runner) +brew install act + +# List workflows +act -l + +# Run push event +act push + +# Run pull request +act pull_request + +# Run release (dry run) +act release -n +``` + +### Test builds manually + +```bash +cd src/amqp-go + +# Test all platforms +for GOOS in darwin linux windows; do + for GOARCH in amd64 arm64; do + echo "Building $GOOS/$GOARCH" + GOOS=$GOOS GOARCH=$GOARCH go build -o /dev/null ./cmd/amqp-server 2>/dev/null && \ + echo " ✓ $GOOS/$GOARCH" || \ + echo " ✗ $GOOS/$GOARCH" + done +done +``` + +## Troubleshooting + +### Release workflow not triggering + +1. Check if release was created (not just a tag) +2. Verify workflow file is in `main` branch +3. Check Actions tab for any errors + +### Build failures + +1. Check Go version compatibility (requires 1.25.1+) +2. Verify all tests pass locally: `go test ./...` +3. Check for platform-specific issues + +### Codecov upload failing + +1. This is non-critical - workflow will continue +2. Add `CODECOV_TOKEN` secret for better reliability +3. Check codecov.io for project setup + +## Customization + +### Change Go version + +Edit all workflow files: + +```yaml +- name: Set up Go + uses: actions/setup-go@v5 + with: + go-version: '1.25.1' # Update this +``` + +### Add new platform + +Edit `release.yml` matrix: + +```yaml +matrix: + include: + # Add new platform + - os: freebsd + arch: amd64 + runner: ubuntu-latest + goos: freebsd + goarch: amd64 +``` + +## Best Practices + +1. **Always test locally** before pushing +2. **Create feature branches** for changes +3. **Wait for CI** before merging PRs +4. **Review coverage reports** to maintain quality +5. **Update workflows** when adding new dependencies +6. **Keep actions up to date** (Dependabot helps) +7. **Monitor workflow runs** for failures + +## Resources + +- [GitHub Actions Documentation](https://docs.github.com/en/actions) +- [Go GitHub Actions](https://github.com/actions/setup-go) +- [CodeQL Documentation](https://codeql.github.com/docs/) + diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml new file mode 100644 index 0000000..ab368e5 --- /dev/null +++ b/.github/workflows/build.yml @@ -0,0 +1,107 @@ +name: Build and Test + +on: + push: + branches: [ main, develop ] + pull_request: + branches: [ main, develop ] + +jobs: + test: + name: Test on Go ${{ matrix.go-version }} + runs-on: ubuntu-latest + strategy: + matrix: + go-version: ['1.25.1'] + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version: ${{ matrix.go-version }} + + - name: Cache Go modules + uses: actions/cache@v4 + with: + path: | + ~/.cache/go-build + ~/go/pkg/mod + key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }} + restore-keys: | + ${{ runner.os }}-go- + + - name: Download dependencies + run: | + cd src/amqp-go + go mod download + + - name: Run tests + run: | + cd src/amqp-go + # Exclude examples directory from tests (contains multiple main() functions) + go list ./... | grep -v '/examples' | xargs go test -v -race -coverprofile=coverage.out -covermode=atomic + + - name: Upload coverage to Codecov + uses: codecov/codecov-action@v4 + with: + files: ./src/amqp-go/coverage.out + flags: unittests + name: codecov-umbrella + fail_ci_if_error: false + + lint: + name: Lint + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version: '1.25.1' + + - name: Run go fmt + run: | + cd src/amqp-go + # Exclude examples and testdata directories + find . -name '*.go' -not -path './examples/*' -not -path './testdata/*' -exec gofmt -l {} + | tee /tmp/fmt-output + if [ -s /tmp/fmt-output ]; then + echo "The following files are not formatted:" + cat /tmp/fmt-output + exit 1 + fi + + - name: Run go vet + run: | + cd src/amqp-go + # Use go list to get all packages except examples and testdata + go list ./... | grep -v '/examples' | grep -v '/testdata' | xargs go vet + + build: + name: Build + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version: '1.25.1' + + - name: Build binary + run: | + cd src/amqp-go + go build -v ./cmd/amqp-server + + - name: Verify build + run: | + cd src/amqp-go + ./amqp-server -version + diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml new file mode 100644 index 0000000..8238463 --- /dev/null +++ b/.github/workflows/codeql.yml @@ -0,0 +1,41 @@ +name: "CodeQL" + +on: + push: + branches: [ "main" ] + pull_request: + branches: [ "main" ] + schedule: + - cron: '0 0 * * 1' # Weekly on Monday + +jobs: + analyze: + name: Analyze + runs-on: ubuntu-latest + permissions: + actions: read + contents: read + security-events: write + + strategy: + fail-fast: false + matrix: + language: [ 'go' ] + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Initialize CodeQL + uses: github/codeql-action/init@v3 + with: + languages: ${{ matrix.language }} + + - name: Autobuild + uses: github/codeql-action/autobuild@v3 + + - name: Perform CodeQL Analysis + uses: github/codeql-action/analyze@v3 + with: + category: "/language:${{matrix.language}}" + diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..e5c86f5 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,185 @@ +name: Release Build + +on: + release: + types: [created, published] + push: + tags: + - 'v*' + workflow_dispatch: + inputs: + version: + description: 'Version tag (e.g., v1.0.0)' + required: true + default: 'v0.0.0-dev' + +permissions: + contents: write + +jobs: + build: + name: Build ${{ matrix.os }} - ${{ matrix.arch }} + runs-on: ${{ matrix.runner }} + strategy: + matrix: + include: + # macOS builds + - os: darwin + arch: amd64 + runner: macos-latest + goos: darwin + goarch: amd64 + - os: darwin + arch: arm64 + runner: macos-latest + goos: darwin + goarch: arm64 + + # Linux builds + - os: linux + arch: amd64 + runner: ubuntu-latest + goos: linux + goarch: amd64 + - os: linux + arch: arm64 + runner: ubuntu-latest + goos: linux + goarch: arm64 + - os: linux + arch: '386' + runner: ubuntu-latest + goos: linux + goarch: '386' + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version: '1.25.1' + + - name: Get version + id: version + run: | + if [ "${{ github.event_name }}" = "release" ]; then + echo "version=${{ github.event.release.tag_name }}" >> $GITHUB_OUTPUT + elif [ "${{ github.event_name }}" = "push" ]; then + echo "version=${{ github.ref_name }}" >> $GITHUB_OUTPUT + else + echo "version=${{ github.event.inputs.version }}" >> $GITHUB_OUTPUT + fi + shell: bash + + - name: Build binary and create package + env: + GOOS: ${{ matrix.goos }} + GOARCH: ${{ matrix.goarch }} + CGO_ENABLED: 0 + run: | + cd src/amqp-go + VERSION="${{ steps.version.outputs.version }}" + BINARY_NAME="amqp-server" + PACKAGE_NAME="amqp-server-${VERSION}-${{ matrix.os }}-${{ matrix.arch }}" + + # Build with version information + go build \ + -ldflags="-s -w -X main.version=${VERSION}" \ + -o "${BINARY_NAME}" \ + ./cmd/amqp-server + + # Create package directory + mkdir -p "${PACKAGE_NAME}" + + # Copy files into package + cp "${BINARY_NAME}" "${PACKAGE_NAME}/" + cp config.sample.yaml "${PACKAGE_NAME}/" + cp README.release.txt "${PACKAGE_NAME}/README.txt" + cp ../../LICENSE "${PACKAGE_NAME}/" + + # Create tar.gz package + tar -czf "${PACKAGE_NAME}.tar.gz" "${PACKAGE_NAME}" + + # Create checksum for the package + if command -v sha256sum &> /dev/null; then + sha256sum "${PACKAGE_NAME}.tar.gz" > "${PACKAGE_NAME}.tar.gz.sha256" + else + shasum -a 256 "${PACKAGE_NAME}.tar.gz" > "${PACKAGE_NAME}.tar.gz.sha256" + fi + + # List package contents + echo "Package contents:" + tar -tzf "${PACKAGE_NAME}.tar.gz" + ls -lh "${PACKAGE_NAME}.tar.gz" + shell: bash + + - name: Upload release package + uses: actions/upload-artifact@v4 + with: + name: amqp-server-${{ matrix.os }}-${{ matrix.arch }} + path: | + src/amqp-go/amqp-server-*.tar.gz* + retention-days: 5 + + release: + name: Create Release Assets + needs: build + runs-on: ubuntu-latest + if: github.event_name == 'release' || github.event_name == 'push' + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Get version tag + id: version + run: | + if [ "${{ github.event_name }}" = "release" ]; then + echo "tag=${{ github.event.release.tag_name }}" >> $GITHUB_OUTPUT + else + echo "tag=${{ github.ref_name }}" >> $GITHUB_OUTPUT + fi + shell: bash + + - name: Download all artifacts + uses: actions/download-artifact@v4 + with: + path: artifacts + + - name: Display structure + run: ls -R artifacts + + - name: Create release if needed + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: | + TAG="${{ steps.version.outputs.tag }}" + # Check if release exists, create if it doesn't + if ! gh release view "$TAG" &>/dev/null; then + echo "Creating release for tag $TAG" + gh release create "$TAG" \ + --title "Release $TAG" \ + --notes "Automated release for $TAG" \ + --draft=false + else + echo "Release $TAG already exists" + fi + shell: bash + + - name: Upload release assets + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: | + TAG="${{ steps.version.outputs.tag }}" + for dir in artifacts/*/; do + for file in "${dir}"*; do + if [ -f "$file" ]; then + echo "Uploading: $file" + gh release upload "$TAG" "$file" --clobber + fi + done + done + shell: bash + diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9cf34de --- /dev/null +++ b/.gitignore @@ -0,0 +1,70 @@ +# Binaries for programs and plugins +*.exe +*.exe~ +*.dll +*.so +*.dylib + +# Test binary, built with `go test -c` +*.test + +# Output of the go coverage tool, specifically when used with LiteIDE +*.out + +# Dependency directories (remove the comment below to include it) +vendor/ + +# Go workspace file +go.work + +# Build artifacts +bin/ +dist/ +/amqp-go +/amqp-server +amqp-server-* + +# IDE +.idea/ +.vscode/ +.cursor/ +*.swp +*.swo +*~ +.DS_Store + +# Logs +*.log +logs/ + +# Storage directories +*.db +*.badger +data/ +amqp-data/ + +# Test files +tests.test +coverage.out +coverage.html + +# PID files +*.pid + +# Configuration overrides +config.local.json +auth.local.json + +# Temporary files +tmp/ +temp/ +*.tmp + +# OS specific +Thumbs.db +.DS_Store + +# Environment files +.env +.env.local +.claude/ diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..49b548c --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,69 @@ +# Changelog + +All notable changes to this project will be documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## [Unreleased] + +### Added +- GitHub Actions workflow for automated multi-platform releases +- Comprehensive build automation for macOS (arm64, amd64), Linux (amd64, arm64, 386) +- Issue and PR templates for better contribution workflow + +## [0.1.0] - TBD + +### Added +- Full AMQP 0.9.1 protocol implementation +- Connection and channel management +- Exchange operations (direct, fanout, topic, headers) +- Queue operations with durability support +- Message publishing and consumption +- Transaction support (tx.select, tx.commit, tx.rollback) +- Multiple storage backends: + - In-memory storage for development + - Badger-based persistent storage +- Authentication framework: + - SASL PLAIN mechanism + - SASL ANONYMOUS mechanism + - File-based authentication with bcrypt +- Prometheus metrics integration +- Comprehensive configuration system +- Builder pattern API for server construction +- Recovery and durability features +- Performance optimizations: + - Buffer pooling (sync.Pool) + - Optimized frame operations + - Zero-allocation hot paths +- Extensive test coverage: + - Protocol compliance tests + - RabbitMQ interoperability tests + - Fuzz testing + - Performance benchmarks + +### Performance +- Memory storage: 3M+ operations/sec +- Message throughput: Up to 3.6 GB/s +- Frame processing: 24-65 million ops/sec +- Protocol operations: 15-140 ns/op + +### Documentation +- Complete README with examples +- Configuration guide +- Authentication setup guide +- Metrics and monitoring guide +- Performance documentation +- RabbitMQ test documentation + +### Infrastructure +- Modular package structure +- Interface-based design +- Comprehensive error handling +- Graceful shutdown support +- PID file management +- Daemon mode support + +[Unreleased]: https://github.com/maxpert/amqp-go/compare/v0.1.0...HEAD +[0.1.0]: https://github.com/maxpert/amqp-go/releases/tag/v0.1.0 + diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..0e62211 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,31 @@ +## Directory Overview +This directory is intended for the development of a Go-based AMQP 0.9.1 server implementation as outlined in the plan.md file. Currently, it contains the project plan and this context file. + +## Project Plan Summary +The project aims to create a Go package `github.com/maxpert/amqp-go` that implements an AMQP 0.9.1 server based on the official specification. It is broken down into several phases: + +1. **Project Setup and Core Protocol Foundation:** Initialize the module, define core data structures, implement binary frame handling, basic TCP server, and connection/channel management. +2. **Exchange and Queue Management:** Implement core exchange types and queue operations. +3. **Message Publishing and Consumption:** Implement basic publishing, consuming, and acknowledgment mechanisms. +4. **Persistence and Reliability:** Integrate a storage backend and implement message durability. +5. **Advanced Features and Security:** Add features like transactions, access control, and TLS/SSL support. +6. **Testing and Refinement:** Comprehensive testing, benchmarking, and documentation. + +## Key Files +- `plan.md`: Contains the detailed plan, phases, and questions for the AMQP server implementation. +- `CLAUDE.md`: This context file for Claude Code. + +## Project Type +This is intended to be a **Go code project** implementing an AMQP 0.9.1 server library. + +## Memories + +- Never write bogus test cases that don't actually test the code +- Follow broadly followed Go community guidelines +- Always have performance considerations when using a particular approach or library +- Always commit changes on every major step, update the corresponding plan file before a commit +- Commit messages should be detailed and opensource standard +- Prefer TDD, if the test doesn't work it can't be a commit! +- When in doubt ask questions, first look internet for what is most obvious answer and then ask question with recommendation +- Always format changes according to go standards before compiling. +- always update plan.md file after you have completed a step. No checkin should go without updated plan.md \ No newline at end of file diff --git a/CLIENT_EXAMPLES.md b/CLIENT_EXAMPLES.md new file mode 100644 index 0000000..d31398b --- /dev/null +++ b/CLIENT_EXAMPLES.md @@ -0,0 +1,963 @@ +# Client Examples + +Complete examples for connecting to StrangeQ from various programming languages. + +## Table of Contents + +- [Python (pika)](#python-pika) +- [Node.js (amqplib)](#nodejs-amqplib) +- [Go (amqp091-go)](#go-amqp091-go) + +--- + +## Python (pika) + +### Installation + +```bash +pip install pika +``` + +### Basic Publisher + +```python +import pika + +def publish_message(): + # Connect to StrangeQ + connection = pika.BlockingConnection( + pika.ConnectionParameters('localhost') + ) + channel = connection.channel() + + # Declare queue + channel.queue_declare(queue='task_queue', durable=True) + + # Publish message + message = "Hello from Python!" + channel.basic_publish( + exchange='', + routing_key='task_queue', + body=message, + properties=pika.BasicProperties( + delivery_mode=2, # make message persistent + ) + ) + + print(f" [x] Sent '{message}'") + connection.close() + +if __name__ == '__main__': + publish_message() +``` + +### Basic Consumer + +```python +import pika + +def callback(ch, method, properties, body): + print(f" [x] Received {body.decode()}") + # Acknowledge the message + ch.basic_ack(delivery_tag=method.delivery_tag) + +def consume_messages(): + connection = pika.BlockingConnection( + pika.ConnectionParameters('localhost') + ) + channel = connection.channel() + + # Declare queue + channel.queue_declare(queue='task_queue', durable=True) + + # Fair dispatch - don't give more than one message at a time + channel.basic_qos(prefetch_count=1) + + # Start consuming + channel.basic_consume( + queue='task_queue', + on_message_callback=callback + ) + + print(' [*] Waiting for messages. To exit press CTRL+C') + channel.start_consuming() + +if __name__ == '__main__': + try: + consume_messages() + except KeyboardInterrupt: + print('Interrupted') +``` + +### Publish/Subscribe with Fanout Exchange + +```python +import pika + +# Publisher +def publish_logs(): + connection = pika.BlockingConnection( + pika.ConnectionParameters('localhost') + ) + channel = connection.channel() + + # Declare fanout exchange + channel.exchange_declare(exchange='logs', exchange_type='fanout') + + message = "Log message" + channel.basic_publish(exchange='logs', routing_key='', body=message) + + print(f" [x] Sent '{message}'") + connection.close() + +# Subscriber +def receive_logs(): + connection = pika.BlockingConnection( + pika.ConnectionParameters('localhost') + ) + channel = connection.channel() + + channel.exchange_declare(exchange='logs', exchange_type='fanout') + + # Create exclusive queue for this consumer + result = channel.queue_declare(queue='', exclusive=True) + queue_name = result.method.queue + + # Bind queue to exchange + channel.queue_bind(exchange='logs', queue=queue_name) + + def callback(ch, method, properties, body): + print(f" [x] {body.decode()}") + + channel.basic_consume( + queue=queue_name, + on_message_callback=callback, + auto_ack=True + ) + + print(' [*] Waiting for logs. To exit press CTRL+C') + channel.start_consuming() +``` + +### Topic Exchange (Routing) + +```python +import pika +import sys + +# Publisher +def publish_to_topic(routing_key, message): + connection = pika.BlockingConnection( + pika.ConnectionParameters('localhost') + ) + channel = connection.channel() + + channel.exchange_declare(exchange='topic_logs', exchange_type='topic') + + channel.basic_publish( + exchange='topic_logs', + routing_key=routing_key, + body=message + ) + + print(f" [x] Sent '{routing_key}':'{message}'") + connection.close() + +# Example: publish_to_topic('kern.critical', 'A critical kernel error') + +# Consumer +def consume_from_topic(binding_keys): + connection = pika.BlockingConnection( + pika.ConnectionParameters('localhost') + ) + channel = connection.channel() + + channel.exchange_declare(exchange='topic_logs', exchange_type='topic') + + result = channel.queue_declare(queue='', exclusive=True) + queue_name = result.method.queue + + # Bind queue with multiple routing patterns + for binding_key in binding_keys: + channel.queue_bind( + exchange='topic_logs', + queue=queue_name, + routing_key=binding_key + ) + + def callback(ch, method, properties, body): + print(f" [x] {method.routing_key}:{body.decode()}") + + channel.basic_consume( + queue=queue_name, + on_message_callback=callback, + auto_ack=True + ) + + print(' [*] Waiting for logs. To exit press CTRL+C') + channel.start_consuming() + +# Example: consume_from_topic(['kern.*', '*.critical']) +``` + +### RPC Pattern + +```python +import pika +import uuid + +# RPC Client +class FibonacciRpcClient: + def __init__(self): + self.connection = pika.BlockingConnection( + pika.ConnectionParameters('localhost') + ) + self.channel = self.connection.channel() + + result = self.channel.queue_declare(queue='', exclusive=True) + self.callback_queue = result.method.queue + + self.channel.basic_consume( + queue=self.callback_queue, + on_message_callback=self.on_response, + auto_ack=True + ) + + self.response = None + self.corr_id = None + + def on_response(self, ch, method, props, body): + if self.corr_id == props.correlation_id: + self.response = body + + def call(self, n): + self.response = None + self.corr_id = str(uuid.uuid4()) + self.channel.basic_publish( + exchange='', + routing_key='rpc_queue', + properties=pika.BasicProperties( + reply_to=self.callback_queue, + correlation_id=self.corr_id, + ), + body=str(n) + ) + + while self.response is None: + self.connection.process_data_events() + + return int(self.response) + +# RPC Server +def fib(n): + if n == 0: + return 0 + elif n == 1: + return 1 + else: + return fib(n-1) + fib(n-2) + +def on_request(ch, method, props, body): + n = int(body) + print(f" [.] fib({n})") + response = fib(n) + + ch.basic_publish( + exchange='', + routing_key=props.reply_to, + properties=pika.BasicProperties(correlation_id=props.correlation_id), + body=str(response) + ) + ch.basic_ack(delivery_tag=method.delivery_tag) + +def start_rpc_server(): + connection = pika.BlockingConnection( + pika.ConnectionParameters('localhost') + ) + channel = connection.channel() + channel.queue_declare(queue='rpc_queue') + channel.basic_qos(prefetch_count=1) + channel.basic_consume(queue='rpc_queue', on_message_callback=on_request) + + print(" [x] Awaiting RPC requests") + channel.start_consuming() +``` + +--- + +## Node.js (amqplib) + +### Installation + +```bash +npm install amqplib +``` + +### Basic Publisher + +```javascript +const amqp = require('amqplib'); + +async function publishMessage() { + try { + const connection = await amqp.connect('amqp://localhost'); + const channel = await connection.createChannel(); + + const queue = 'task_queue'; + const message = 'Hello from Node.js!'; + + await channel.assertQueue(queue, { + durable: true + }); + + channel.sendToQueue(queue, Buffer.from(message), { + persistent: true + }); + + console.log(` [x] Sent '${message}'`); + + setTimeout(() => { + connection.close(); + }, 500); + } catch (error) { + console.error(error); + } +} + +publishMessage(); +``` + +### Basic Consumer + +```javascript +const amqp = require('amqplib'); + +async function consumeMessages() { + try { + const connection = await amqp.connect('amqp://localhost'); + const channel = await connection.createChannel(); + + const queue = 'task_queue'; + + await channel.assertQueue(queue, { + durable: true + }); + + // Fair dispatch + channel.prefetch(1); + + console.log(' [*] Waiting for messages. To exit press CTRL+C'); + + channel.consume(queue, (msg) => { + const content = msg.content.toString(); + console.log(` [x] Received '${content}'`); + + // Simulate work + setTimeout(() => { + console.log(' [x] Done'); + channel.ack(msg); + }, 1000); + }); + } catch (error) { + console.error(error); + } +} + +consumeMessages(); +``` + +### Publish/Subscribe with Fanout Exchange + +```javascript +const amqp = require('amqplib'); + +// Publisher +async function publishLogs() { + const connection = await amqp.connect('amqp://localhost'); + const channel = await connection.createChannel(); + + const exchange = 'logs'; + const message = 'Log message from Node.js'; + + await channel.assertExchange(exchange, 'fanout', { + durable: false + }); + + channel.publish(exchange, '', Buffer.from(message)); + console.log(` [x] Sent '${message}'`); + + setTimeout(() => { + connection.close(); + }, 500); +} + +// Subscriber +async function receiveLogs() { + const connection = await amqp.connect('amqp://localhost'); + const channel = await connection.createChannel(); + + const exchange = 'logs'; + + await channel.assertExchange(exchange, 'fanout', { + durable: false + }); + + const q = await channel.assertQueue('', { + exclusive: true + }); + + console.log(' [*] Waiting for logs. To exit press CTRL+C'); + + channel.bindQueue(q.queue, exchange, ''); + + channel.consume(q.queue, (msg) => { + console.log(` [x] ${msg.content.toString()}`); + }, { + noAck: true + }); +} +``` + +### Topic Exchange (Routing) + +```javascript +const amqp = require('amqplib'); + +// Publisher +async function publishToTopic(routingKey, message) { + const connection = await amqp.connect('amqp://localhost'); + const channel = await connection.createChannel(); + + const exchange = 'topic_logs'; + + await channel.assertExchange(exchange, 'topic', { + durable: false + }); + + channel.publish(exchange, routingKey, Buffer.from(message)); + console.log(` [x] Sent '${routingKey}':'${message}'`); + + setTimeout(() => { + connection.close(); + }, 500); +} + +// Example: publishToTopic('kern.critical', 'A critical kernel error'); + +// Consumer +async function consumeFromTopic(bindingKeys) { + const connection = await amqp.connect('amqp://localhost'); + const channel = await connection.createChannel(); + + const exchange = 'topic_logs'; + + await channel.assertExchange(exchange, 'topic', { + durable: false + }); + + const q = await channel.assertQueue('', { + exclusive: true + }); + + console.log(' [*] Waiting for logs. To exit press CTRL+C'); + + bindingKeys.forEach((key) => { + channel.bindQueue(q.queue, exchange, key); + }); + + channel.consume(q.queue, (msg) => { + console.log(` [x] ${msg.fields.routingKey}:${msg.content.toString()}`); + }, { + noAck: true + }); +} + +// Example: consumeFromTopic(['kern.*', '*.critical']); +``` + +### RPC Pattern + +```javascript +const amqp = require('amqplib'); +const { v4: uuidv4 } = require('uuid'); + +// RPC Client +class FibonacciRpcClient { + async connect() { + this.connection = await amqp.connect('amqp://localhost'); + this.channel = await this.connection.createChannel(); + + const { queue } = await this.channel.assertQueue('', { + exclusive: true + }); + + this.callbackQueue = queue; + + this.channel.consume(this.callbackQueue, (msg) => { + if (msg.properties.correlationId === this.correlationId) { + this.response = msg.content.toString(); + } + }, { noAck: true }); + } + + async call(n) { + this.correlationId = uuidv4(); + this.response = null; + + this.channel.sendToQueue('rpc_queue', Buffer.from(n.toString()), { + correlationId: this.correlationId, + replyTo: this.callbackQueue + }); + + // Wait for response + return new Promise((resolve) => { + const interval = setInterval(() => { + if (this.response !== null) { + clearInterval(interval); + resolve(parseInt(this.response)); + } + }, 100); + }); + } + + async close() { + await this.connection.close(); + } +} + +// RPC Server +function fib(n) { + if (n === 0) return 0; + if (n === 1) return 1; + return fib(n - 1) + fib(n - 2); +} + +async function startRpcServer() { + const connection = await amqp.connect('amqp://localhost'); + const channel = await connection.createChannel(); + + await channel.assertQueue('rpc_queue', { + durable: false + }); + + channel.prefetch(1); + + console.log(' [x] Awaiting RPC requests'); + + channel.consume('rpc_queue', (msg) => { + const n = parseInt(msg.content.toString()); + + console.log(` [.] fib(${n})`); + const result = fib(n); + + channel.sendToQueue(msg.properties.replyTo, Buffer.from(result.toString()), { + correlationId: msg.properties.correlationId + }); + + channel.ack(msg); + }); +} +``` + +--- + +## Go (amqp091-go) + +### Installation + +```bash +go get github.com/rabbitmq/amqp091-go +``` + +### Basic Publisher + +```go +package main + +import ( + "log" + amqp "github.com/rabbitmq/amqp091-go" +) + +func main() { + conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/") + if err != nil { + log.Fatal(err) + } + defer conn.Close() + + ch, err := conn.Channel() + if err != nil { + log.Fatal(err) + } + defer ch.Close() + + q, err := ch.QueueDeclare( + "task_queue", // name + true, // durable + false, // delete when unused + false, // exclusive + false, // no-wait + nil, // arguments + ) + if err != nil { + log.Fatal(err) + } + + body := "Hello from Go!" + err = ch.Publish( + "", // exchange + q.Name, // routing key + false, // mandatory + false, // immediate + amqp.Publishing{ + DeliveryMode: amqp.Persistent, + ContentType: "text/plain", + Body: []byte(body), + }, + ) + if err != nil { + log.Fatal(err) + } + + log.Printf(" [x] Sent '%s'", body) +} +``` + +### Basic Consumer + +```go +package main + +import ( + "log" + amqp "github.com/rabbitmq/amqp091-go" +) + +func main() { + conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/") + if err != nil { + log.Fatal(err) + } + defer conn.Close() + + ch, err := conn.Channel() + if err != nil { + log.Fatal(err) + } + defer ch.Close() + + q, err := ch.QueueDeclare( + "task_queue", + true, + false, + false, + false, + nil, + ) + if err != nil { + log.Fatal(err) + } + + err = ch.Qos( + 1, // prefetch count + 0, // prefetch size + false, // global + ) + if err != nil { + log.Fatal(err) + } + + msgs, err := ch.Consume( + q.Name, + "", + false, // auto-ack + false, + false, + false, + nil, + ) + if err != nil { + log.Fatal(err) + } + + forever := make(chan bool) + + go func() { + for d := range msgs { + log.Printf(" [x] Received %s", d.Body) + d.Ack(false) + } + }() + + log.Printf(" [*] Waiting for messages. To exit press CTRL+C") + <-forever +} +``` + +### Publish/Subscribe with Fanout Exchange + +```go +package main + +import ( + "log" + amqp "github.com/rabbitmq/amqp091-go" +) + +// Publisher +func publishLogs() { + conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/") + if err != nil { + log.Fatal(err) + } + defer conn.Close() + + ch, err := conn.Channel() + if err != nil { + log.Fatal(err) + } + defer ch.Close() + + err = ch.ExchangeDeclare( + "logs", // name + "fanout", // type + true, // durable + false, // auto-deleted + false, // internal + false, // no-wait + nil, // arguments + ) + if err != nil { + log.Fatal(err) + } + + body := "Log message from Go" + err = ch.Publish( + "logs", // exchange + "", // routing key + false, + false, + amqp.Publishing{ + ContentType: "text/plain", + Body: []byte(body), + }, + ) + if err != nil { + log.Fatal(err) + } + + log.Printf(" [x] Sent '%s'", body) +} + +// Subscriber +func receiveLogs() { + conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/") + if err != nil { + log.Fatal(err) + } + defer conn.Close() + + ch, err := conn.Channel() + if err != nil { + log.Fatal(err) + } + defer ch.Close() + + err = ch.ExchangeDeclare( + "logs", + "fanout", + true, + false, + false, + false, + nil, + ) + if err != nil { + log.Fatal(err) + } + + q, err := ch.QueueDeclare( + "", // name + false, // durable + false, // delete when unused + true, // exclusive + false, // no-wait + nil, // arguments + ) + if err != nil { + log.Fatal(err) + } + + err = ch.QueueBind( + q.Name, // queue name + "", // routing key + "logs", // exchange + false, + nil, + ) + if err != nil { + log.Fatal(err) + } + + msgs, err := ch.Consume( + q.Name, + "", + true, + false, + false, + false, + nil, + ) + if err != nil { + log.Fatal(err) + } + + forever := make(chan bool) + + go func() { + for d := range msgs { + log.Printf(" [x] %s", d.Body) + } + }() + + log.Printf(" [*] Waiting for logs. To exit press CTRL+C") + <-forever +} +``` + +### Topic Exchange (Routing) + +```go +package main + +import ( + "log" + amqp "github.com/rabbitmq/amqp091-go" +) + +func publishToTopic(routingKey string, message string) { + conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/") + if err != nil { + log.Fatal(err) + } + defer conn.Close() + + ch, err := conn.Channel() + if err != nil { + log.Fatal(err) + } + defer ch.Close() + + err = ch.ExchangeDeclare( + "topic_logs", + "topic", + true, + false, + false, + false, + nil, + ) + if err != nil { + log.Fatal(err) + } + + err = ch.Publish( + "topic_logs", + routingKey, + false, + false, + amqp.Publishing{ + ContentType: "text/plain", + Body: []byte(message), + }, + ) + if err != nil { + log.Fatal(err) + } + + log.Printf(" [x] Sent '%s':'%s'", routingKey, message) +} + +func consumeFromTopic(bindingKeys []string) { + conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/") + if err != nil { + log.Fatal(err) + } + defer conn.Close() + + ch, err := conn.Channel() + if err != nil { + log.Fatal(err) + } + defer ch.Close() + + err = ch.ExchangeDeclare( + "topic_logs", + "topic", + true, + false, + false, + false, + nil, + ) + if err != nil { + log.Fatal(err) + } + + q, err := ch.QueueDeclare( + "", + false, + false, + true, + false, + nil, + ) + if err != nil { + log.Fatal(err) + } + + for _, key := range bindingKeys { + err = ch.QueueBind( + q.Name, + key, + "topic_logs", + false, + nil, + ) + if err != nil { + log.Fatal(err) + } + } + + msgs, err := ch.Consume( + q.Name, + "", + true, + false, + false, + false, + nil, + ) + if err != nil { + log.Fatal(err) + } + + forever := make(chan bool) + + go func() { + for d := range msgs { + log.Printf(" [x] %s:%s", d.RoutingKey, d.Body) + } + }() + + log.Printf(" [*] Waiting for logs. To exit press CTRL+C") + <-forever +} +``` + +## Next Steps + +- See [QUICKSTART.md](QUICKSTART.md) for server installation and configuration +- Check [CONTRIBUTING.md](CONTRIBUTING.md) for development guidelines +- Review [docs/plan.md](docs/plan.md) for implementation details diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..d09aaf9 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,300 @@ +# Contributing to AMQP-Go + +Thank you for your interest in contributing to AMQP-Go! This document provides guidelines and instructions for contributing. + +## Table of Contents +- [Code of Conduct](#code-of-conduct) +- [Getting Started](#getting-started) +- [Development Setup](#development-setup) +- [Making Changes](#making-changes) +- [Testing](#testing) +- [Submitting Changes](#submitting-changes) +- [Code Style](#code-style) +- [Commit Messages](#commit-messages) + +## Code of Conduct + +Please be respectful and constructive in all interactions. We aim to foster an open and welcoming environment. + +## Getting Started + +1. **Fork the repository** on GitHub +2. **Clone your fork** locally: + ```bash + git clone https://github.com/YOUR-USERNAME/strangeq.git + cd strangeq/src/amqp-go + ``` +3. **Add upstream remote**: + ```bash + git remote add upstream https://github.com/maxpert/strangeq.git + ``` + +## Development Setup + +### Prerequisites +- Go 1.25.1 or later +- Git +- Make (optional, for build automation) + +### Install Dependencies +```bash +cd src/amqp-go +go mod download +``` + +### Build the Project +```bash +go build ./cmd/amqp-server +``` + +### Run Tests +```bash +# Run all tests +go test ./... + +# Run tests with coverage +go test -coverprofile=coverage.out ./... +go tool cover -html=coverage.out + +# Run specific package tests +go test ./protocol/... +``` + +### Run Benchmarks +```bash +go test -bench=. -benchmem ./protocol/ +go test -bench=. -benchmem ./benchmarks/ +``` + +## Making Changes + +1. **Create a feature branch**: + ```bash + git checkout -b feature/your-feature-name + # or + git checkout -b fix/issue-description + ``` + +2. **Make your changes**: + - Write clear, concise code + - Add tests for new functionality + - Update documentation as needed + - Follow the existing code style + +3. **Test your changes**: + ```bash + # Run tests + go test ./... + + # Run linter + golangci-lint run + + # Format code + go fmt ./... + ``` + +4. **Commit your changes** (see [Commit Messages](#commit-messages)) + +## Testing + +### Test Requirements +- All new features must include tests +- Bug fixes should include regression tests +- Maintain or improve code coverage +- Tests should be deterministic and not flaky + +### Test Categories + +**Unit Tests** +```bash +go test ./broker/ +go test ./protocol/ +go test ./storage/ +``` + +**Integration Tests** +```bash +go test ./tests/ +go test -tags=integration ./... +``` + +**Performance Benchmarks** +```bash +go test -bench=. -benchmem ./benchmarks/ +``` + +**Fuzz Tests** +```bash +go test -fuzz=FuzzFrameMarshaling -fuzztime=30s ./protocol/ +``` + +### Writing Tests +- Use table-driven tests when appropriate +- Use testify for assertions +- Clean up resources (defer cleanup functions) +- Use meaningful test names that describe what's being tested + +Example: +```go +func TestQueueDeclare(t *testing.T) { + tests := []struct { + name string + queueName string + durable bool + wantErr bool + }{ + {"valid queue", "test-queue", true, false}, + {"empty name", "", false, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // test implementation + }) + } +} +``` + +## Submitting Changes + +1. **Push to your fork**: + ```bash + git push origin feature/your-feature-name + ``` + +2. **Create a Pull Request**: + - Go to the repository on GitHub + - Click "New Pull Request" + - Select your branch + - Fill out the PR template completely + - Link related issues + +3. **Address Review Comments**: + - Respond to all review comments + - Make requested changes + - Push updates to the same branch + +4. **Keep PR Updated**: + ```bash + # Sync with upstream + git fetch upstream + git rebase upstream/main + git push -f origin feature/your-feature-name + ``` + +## Code Style + +### General Guidelines +- Follow standard Go conventions +- Use `gofmt` for formatting +- Run `golangci-lint` before committing +- Write self-documenting code with clear variable names +- Add comments for complex logic +- Keep functions focused and concise + +### Package Documentation +- Every package should have a package-level comment +- Export functions and types should have comments +- Comments should start with the name of the thing being described + +Example: +```go +// Package broker provides message routing and delivery functionality +// for the AMQP server. +package broker + +// DeclareQueue creates a new queue or returns an existing one. +// It returns an error if the queue parameters conflict with an existing queue. +func DeclareQueue(name string, durable bool) error { + // implementation +} +``` + +### Error Handling +- Return errors, don't panic (except in truly exceptional cases) +- Use custom error types from the `errors` package +- Wrap errors with context: `fmt.Errorf("operation failed: %w", err)` +- Check all errors + +### Concurrency +- Use mutexes appropriately (prefer RWMutex when reads dominate) +- Document goroutine lifecycles +- Use channels for communication between goroutines +- Avoid goroutine leaks (ensure all goroutines can exit) + +## Commit Messages + +### Format +``` +(): + + + +