Browse Source
Merge pull request #93 from aaqaishtyaq/add-pub-ctx
Remove deprecated 'Publish' methodpull/104/head
Lane Wagner
3 years ago
committed by
GitHub
GitHub
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
34 changed files with 96 additions and 8767 deletions
Split View
Diff Options
-
+3 -1examples/logger/main.go
-
+5 -2examples/multipublisher/main.go
-
+3 -1examples/publisher/main.go
-
+1 -1go.mod
-
+40 -2go.sum
-
+25 -1internal/channelmanager/safe_wraps.go
-
+19 -3publish.go
-
+0 -6vendor/github.com/rabbitmq/amqp091-go/.gitignore
-
+0 -23vendor/github.com/rabbitmq/amqp091-go/.travis.yml
-
+0 -77vendor/github.com/rabbitmq/amqp091-go/CODE_OF_CONDUCT.md
-
+0 -37vendor/github.com/rabbitmq/amqp091-go/CONTRIBUTING.md
-
+0 -25vendor/github.com/rabbitmq/amqp091-go/LICENSE
-
+0 -17vendor/github.com/rabbitmq/amqp091-go/Makefile
-
+0 -110vendor/github.com/rabbitmq/amqp091-go/README.md
-
+0 -111vendor/github.com/rabbitmq/amqp091-go/allocator.go
-
+0 -83vendor/github.com/rabbitmq/amqp091-go/auth.go
-
+0 -159vendor/github.com/rabbitmq/amqp091-go/certs.sh
-
+0 -4vendor/github.com/rabbitmq/amqp091-go/change_version.sh
-
+0 -1622vendor/github.com/rabbitmq/amqp091-go/channel.go
-
+0 -180vendor/github.com/rabbitmq/amqp091-go/confirms.go
-
+0 -875vendor/github.com/rabbitmq/amqp091-go/connection.go
-
+0 -142vendor/github.com/rabbitmq/amqp091-go/consumers.go
-
+0 -173vendor/github.com/rabbitmq/amqp091-go/delivery.go
-
+0 -146vendor/github.com/rabbitmq/amqp091-go/doc.go
-
+0 -23vendor/github.com/rabbitmq/amqp091-go/fuzz.go
-
+0 -2vendor/github.com/rabbitmq/amqp091-go/gen.sh
-
+0 -67vendor/github.com/rabbitmq/amqp091-go/pre-commit
-
+0 -464vendor/github.com/rabbitmq/amqp091-go/read.go
-
+0 -64vendor/github.com/rabbitmq/amqp091-go/return.go
-
+0 -3306vendor/github.com/rabbitmq/amqp091-go/spec091.go
-
+0 -439vendor/github.com/rabbitmq/amqp091-go/types.go
-
+0 -176vendor/github.com/rabbitmq/amqp091-go/uri.go
-
+0 -422vendor/github.com/rabbitmq/amqp091-go/write.go
-
+0 -3vendor/modules.txt
+ 3
- 1
examples/logger/main.go
View File
+ 5
- 2
examples/multipublisher/main.go
View File
+ 3
- 1
examples/publisher/main.go
View File
+ 1
- 1
go.mod
View File
+ 40
- 2
go.sum
View File
| @ -1,2 +1,40 @@ | |||
| github.com/rabbitmq/amqp091-go v1.3.4 h1:tXuIslN1nhDqs2t6Jrz3BAoqvt4qIZzxvdbdcxWtHYU= | |||
| github.com/rabbitmq/amqp091-go v1.3.4/go.mod h1:ogQDLSOACsLPsIq0NpbtiifNZi2YOz0VTJ0kHRghqbM= | |||
| github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= | |||
| github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= | |||
| github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= | |||
| github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= | |||
| github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= | |||
| github.com/rabbitmq/amqp091-go v1.5.0 h1:VouyHPBu1CrKyJVfteGknGOGCzmOz0zcv/tONLkb7rg= | |||
| github.com/rabbitmq/amqp091-go v1.5.0/go.mod h1:JsV0ofX5f1nwOGafb8L5rBItt9GyhfQfcJj+oyz0dGg= | |||
| github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= | |||
| github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= | |||
| github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= | |||
| go.uber.org/goleak v1.1.12 h1:gZAh5/EyT/HQwlpkCy6wTpqfH9H8Lz8zbm3dZh+OyzA= | |||
| go.uber.org/goleak v1.1.12/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ= | |||
| golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= | |||
| golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= | |||
| golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= | |||
| golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= | |||
| golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= | |||
| golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= | |||
| golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= | |||
| golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= | |||
| golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= | |||
| golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= | |||
| golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= | |||
| golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= | |||
| golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= | |||
| golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= | |||
| golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= | |||
| golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= | |||
| golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= | |||
| golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= | |||
| golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= | |||
| golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= | |||
| golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= | |||
| golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= | |||
| golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= | |||
| golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= | |||
| golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= | |||
| gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= | |||
| gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= | |||
| gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= | |||
+ 25
- 1
internal/channelmanager/safe_wraps.go
View File
+ 19
- 3
publish.go
View File
+ 0
- 6
vendor/github.com/rabbitmq/amqp091-go/.gitignore
View File
| @ -1,6 +0,0 @@ | |||
| certs/* | |||
| spec/spec | |||
| examples/simple-consumer/simple-consumer | |||
| examples/simple-producer/simple-producer | |||
| .idea/ | |||
+ 0
- 23
vendor/github.com/rabbitmq/amqp091-go/.travis.yml
View File
| @ -1,23 +0,0 @@ | |||
| language: go | |||
| go: | |||
| - 1.17.x | |||
| - 1.16.x | |||
| addons: | |||
| apt: | |||
| packages: | |||
| - rabbitmq-server | |||
| services: | |||
| - rabbitmq | |||
| env: | |||
| - GO111MODULE=on AMQP_URL=amqp://guest:guest@127.0.0.1:5672/ | |||
| before_install: | |||
| - go get -v golang.org/x/lint/golint | |||
| script: | |||
| - ./pre-commit | |||
| - go test -cpu=1,2 -v -tags integration ./... | |||
+ 0
- 77
vendor/github.com/rabbitmq/amqp091-go/CODE_OF_CONDUCT.md
View File
| @ -1,77 +0,0 @@ | |||
| # Contributor Covenant Code of Conduct | |||
| ## Our Pledge | |||
| In the interest of fostering an open and welcoming environment, we as | |||
| contributors and maintainers pledge to making participation in RabbitMQ Operator project and | |||
| our community a harassment-free experience for everyone, regardless of age, body | |||
| size, disability, ethnicity, sex characteristics, gender identity and expression, | |||
| level of experience, education, socio-economic status, nationality, personal | |||
| appearance, race, religion, or sexual identity and orientation. | |||
| ## Our Standards | |||
| Examples of behavior that contributes to creating a positive environment | |||
| include: | |||
| * Using welcoming and inclusive language | |||
| * Being respectful of differing viewpoints and experiences | |||
| * Gracefully accepting constructive criticism | |||
| * Focusing on what is best for the community | |||
| * Showing empathy towards other community members | |||
| Examples of unacceptable behavior by participants include: | |||
| * The use of sexualized language or imagery and unwelcome sexual attention or | |||
| advances | |||
| * Trolling, insulting/derogatory comments, and personal or political attacks | |||
| * Public or private harassment | |||
| * Publishing others' private information, such as a physical or electronic | |||
| address, without explicit permission | |||
| * Other conduct which could reasonably be considered inappropriate in a | |||
| professional setting | |||
| ## Our Responsibilities | |||
| Project maintainers are responsible for clarifying the standards of acceptable | |||
| behavior and are expected to take appropriate and fair corrective action in | |||
| response to any instances of unacceptable behavior. | |||
| Project maintainers have the right and responsibility to remove, edit, or | |||
| reject comments, commits, code, wiki edits, issues, and other contributions | |||
| that are not aligned to this Code of Conduct, or to ban temporarily or | |||
| permanently any contributor for other behaviors that they deem inappropriate, | |||
| threatening, offensive, or harmful. | |||
| ## Scope | |||
| This Code of Conduct applies both within project spaces and in public spaces | |||
| when an individual is representing the project or its community. Examples of | |||
| representing a project or community include using an official project e-mail | |||
| address, posting via an official social media account, or acting as an appointed | |||
| representative at an online or offline event. Representation of a project may be | |||
| further defined and clarified by project maintainers. | |||
| ## Enforcement | |||
| Instances of abusive, harassing, or otherwise unacceptable behavior may be | |||
| reported by contacting the project team at oss-coc@vmware.com. All | |||
| complaints will be reviewed and investigated and will result in a response that | |||
| is deemed necessary and appropriate to the circumstances. The project team is | |||
| obligated to maintain confidentiality with regard to the reporter of an incident. | |||
| Further details of specific enforcement policies may be posted separately. | |||
| Project maintainers who do not follow or enforce the Code of Conduct in good | |||
| faith may face temporary or permanent repercussions as determined by other | |||
| members of the project's leadership. | |||
| ## Attribution | |||
| This Code of Conduct is adapted from the [Contributor Covenant][homepage], version 1.4, | |||
| available at https://www.contributor-covenant.org/version/1/4/code-of-conduct.html | |||
| [homepage]: https://www.contributor-covenant.org | |||
| For answers to common questions about this code of conduct, see | |||
| https://www.contributor-covenant.org/faq | |||
+ 0
- 37
vendor/github.com/rabbitmq/amqp091-go/CONTRIBUTING.md
View File
| @ -1,37 +0,0 @@ | |||
| ## Prerequisites | |||
| 1. Go: [https://golang.org/dl/](https://golang.org/dl/) | |||
| 1. Golint `go get -u -v github.com/golang/lint/golint` | |||
| ## Contributing | |||
| The workflow is pretty standard: | |||
| 1. Fork this repository, **github.com/rabbitmq/amqp091-go** | |||
| 1. Add the pre-commit hook: `ln -s ../../pre-commit .git/hooks/pre-commit` | |||
| 1. Create your feature branch (`git checkout -b my-new-feature`) | |||
| 1. Run integration tests (see below) | |||
| 1. **Implement tests** | |||
| 1. Implement fixs | |||
| 1. Commit your changes (`git commit -am 'Add some feature'`) | |||
| 1. Push to a branch (`git push -u origin my-new-feature`) | |||
| 1. Submit a pull request | |||
| ## Running Tests | |||
| The test suite assumes that: | |||
| * A RabbitMQ node is running on localhost with all defaults: [https://www.rabbitmq.com/download.html](https://www.rabbitmq.com/download.html) | |||
| * `AMQP_URL` is exported to `amqp://guest:guest@127.0.0.1:5672/` | |||
| ### Integration Tests | |||
| After starting a local RabbitMQ, run integration tests with the following: | |||
| ``` shell | |||
| env AMQP_URL=amqp://guest:guest@127.0.0.1:5672/ go test -v -tags integration | |||
| ``` | |||
| All integration tests should use the `integrationConnection(...)` test | |||
| helpers defined in `integration_test.go` to setup the integration environment | |||
| and logging. | |||
+ 0
- 25
vendor/github.com/rabbitmq/amqp091-go/LICENSE
View File
| @ -1,25 +0,0 @@ | |||
| AMQP 0-9-1 Go Client | |||
| Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| Redistribution and use in source and binary forms, with or without | |||
| modification, are permitted provided that the following conditions are met: | |||
| Redistributions of source code must retain the above copyright notice, this | |||
| list of conditions and the following disclaimer. | |||
| Redistributions in binary form must reproduce the above copyright notice, this | |||
| list of conditions and the following disclaimer in the documentation and/or | |||
| other materials provided with the distribution. | |||
| THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND | |||
| ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED | |||
| WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE | |||
| DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE | |||
| FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL | |||
| DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR | |||
| SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER | |||
| CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, | |||
| OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE | |||
| OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. | |||
+ 0
- 17
vendor/github.com/rabbitmq/amqp091-go/Makefile
View File
| @ -1,17 +0,0 @@ | |||
| .DEFAULT_GOAL := list | |||
| # Insert a comment starting with '##' after a target, and it will be printed by 'make' and 'make list' | |||
| list: ## list Makefile targets | |||
| @echo "The most used targets: \n" | |||
| @grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-20s\033[0m %s\n", $$1, $$2}' | |||
| fmt: ## Run go fmt against code | |||
| go fmt ./... | |||
| vet: ## Run go vet against code | |||
| go vet ./... | |||
| tests: ## Run all tests and requires a running rabbitmq-server | |||
| env AMQP_URL=amqp://guest:guest@127.0.0.1:5672/ go test -v -tags integration | |||
+ 0
- 110
vendor/github.com/rabbitmq/amqp091-go/README.md
View File
| @ -1,110 +0,0 @@ | |||
| # Go RabbitMQ Client Library | |||
| [](https://pkg.go.dev/github.com/rabbitmq/amqp091-go) | |||
| This is a Go AMQP 0.9.1 client maintained by the [RabbitMQ core team](https://github.com/rabbitmq). | |||
| It was [originally developed by Sean Treadway](https://github.com/streadway/amqp). | |||
| ## Differences from streadway/amqp | |||
| Some things are different compared to the original client, | |||
| others haven't changed. | |||
| ### Package Name | |||
| This library uses a different package name. If moving from `streadway/amqp`, | |||
| using an alias may reduce the number of changes needed: | |||
| ``` go | |||
| amqp "github.com/rabbitmq/amqp091-go" | |||
| ``` | |||
| ### License | |||
| This client uses the same 2-clause BSD license as the original project. | |||
| ### Public API Evolution | |||
| This client retains key API elements as practically possible. | |||
| It is, however, open to reasonable breaking public API changes suggested by the community. | |||
| We don't have the "no breaking public API changes ever" rule and fully recognize | |||
| that a good client API evolves over time. | |||
| ## Project Maturity | |||
| This project is based on a mature Go client that's been around for over a decade. | |||
| We expect this client to undergo moderate breaking public API changes in 2021. | |||
| Major and minor versions will be updated accordingly. | |||
| ## Supported Go Versions | |||
| This client supports two most recent Go release series. | |||
| ## Supported RabbitMQ Versions | |||
| This project supports RabbitMQ versions starting with `2.0` but primarily tested | |||
| against [currently supported RabbitMQ release series](https://www.rabbitmq.com/versions.html). | |||
| Some features and behaviours may be server version-specific. | |||
| ## Goals | |||
| Provide a functional interface that closely represents the AMQP 0.9.1 model | |||
| targeted to RabbitMQ as a server. This includes the minimum necessary to | |||
| interact the semantics of the protocol. | |||
| ## Non-goals | |||
| Things not intended to be supported. | |||
| * Auto reconnect and re-synchronization of client and server topologies. | |||
| * Reconnection would require understanding the error paths when the | |||
| topology cannot be declared on reconnect. This would require a new set | |||
| of types and code paths that are best suited at the call-site of this | |||
| package. AMQP has a dynamic topology that needs all peers to agree. If | |||
| this doesn't happen, the behavior is undefined. Instead of producing a | |||
| possible interface with undefined behavior, this package is designed to | |||
| be simple for the caller to implement the necessary connection-time | |||
| topology declaration so that reconnection is trivial and encapsulated in | |||
| the caller's application code. | |||
| * AMQP Protocol negotiation for forward or backward compatibility. | |||
| * 0.9.1 is stable and widely deployed. Versions 0.10 and 1.0 are divergent | |||
| specifications that change the semantics and wire format of the protocol. | |||
| We will accept patches for other protocol support but have no plans for | |||
| implementation ourselves. | |||
| * Anything other than PLAIN and EXTERNAL authentication mechanisms. | |||
| * Keeping the mechanisms interface modular makes it possible to extend | |||
| outside of this package. If other mechanisms prove to be popular, then | |||
| we would accept patches to include them in this package. | |||
| ## Usage | |||
| See the 'examples' subdirectory for simple producers and consumers executables. | |||
| If you have a use-case in mind which isn't well-represented by the examples, | |||
| please file an issue. | |||
| ## Documentation | |||
| * [Godoc API reference](http://godoc.org/github.com/rabbitmq/amqp091-go) | |||
| * [RabbitMQ tutorials in Go](https://github.com/rabbitmq/rabbitmq-tutorials/tree/master/go) currently use a different client. | |||
| They will be switched to use this client eventually | |||
| ## Contributing | |||
| Pull requests are very much welcomed. Create your pull request on a non-main | |||
| branch, make sure a test or example is included that covers your change, and | |||
| your commits represent coherent changes that include a reason for the change. | |||
| To run the integration tests, make sure you have RabbitMQ running on any host, | |||
| export the environment variable `AMQP_URL=amqp://host/` and run `go test -tags | |||
| integration`. TravisCI will also run the integration tests. | |||
| ## License | |||
| BSD 2 clause - see LICENSE for more details. | |||
+ 0
- 111
vendor/github.com/rabbitmq/amqp091-go/allocator.go
View File
| @ -1,111 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| package amqp091 | |||
| import ( | |||
| "bytes" | |||
| "fmt" | |||
| "math/big" | |||
| ) | |||
| const ( | |||
| free = 0 | |||
| allocated = 1 | |||
| ) | |||
| // allocator maintains a bitset of allocated numbers. | |||
| type allocator struct { | |||
| pool *big.Int | |||
| last int | |||
| low int | |||
| high int | |||
| } | |||
| // NewAllocator reserves and frees integers out of a range between low and | |||
| // high. | |||
| // | |||
| // O(N) worst case space used, where N is maximum allocated, divided by | |||
| // sizeof(big.Word) | |||
| func newAllocator(low, high int) *allocator { | |||
| return &allocator{ | |||
| pool: big.NewInt(0), | |||
| last: low, | |||
| low: low, | |||
| high: high, | |||
| } | |||
| } | |||
| // String returns a string describing the contents of the allocator like | |||
| // "allocator[low..high] reserved..until" | |||
| // | |||
| // O(N) where N is high-low | |||
| func (a allocator) String() string { | |||
| b := &bytes.Buffer{} | |||
| fmt.Fprintf(b, "allocator[%d..%d]", a.low, a.high) | |||
| for low := a.low; low <= a.high; low++ { | |||
| high := low | |||
| for a.reserved(high) && high <= a.high { | |||
| high++ | |||
| } | |||
| if high > low+1 { | |||
| fmt.Fprintf(b, " %d..%d", low, high-1) | |||
| } else if high > low { | |||
| fmt.Fprintf(b, " %d", high-1) | |||
| } | |||
| low = high | |||
| } | |||
| return b.String() | |||
| } | |||
| // Next reserves and returns the next available number out of the range between | |||
| // low and high. If no number is available, false is returned. | |||
| // | |||
| // O(N) worst case runtime where N is allocated, but usually O(1) due to a | |||
| // rolling index into the oldest allocation. | |||
| func (a *allocator) next() (int, bool) { | |||
| wrapped := a.last | |||
| // Find trailing bit | |||
| for ; a.last <= a.high; a.last++ { | |||
| if a.reserve(a.last) { | |||
| return a.last, true | |||
| } | |||
| } | |||
| // Find preceding free'd pool | |||
| a.last = a.low | |||
| for ; a.last < wrapped; a.last++ { | |||
| if a.reserve(a.last) { | |||
| return a.last, true | |||
| } | |||
| } | |||
| return 0, false | |||
| } | |||
| // reserve claims the bit if it is not already claimed, returning true if | |||
| // successfully claimed. | |||
| func (a *allocator) reserve(n int) bool { | |||
| if a.reserved(n) { | |||
| return false | |||
| } | |||
| a.pool.SetBit(a.pool, n-a.low, allocated) | |||
| return true | |||
| } | |||
| // reserved returns true if the integer has been allocated | |||
| func (a *allocator) reserved(n int) bool { | |||
| return a.pool.Bit(n-a.low) == allocated | |||
| } | |||
| // release frees the use of the number for another allocation | |||
| func (a *allocator) release(n int) { | |||
| a.pool.SetBit(a.pool, n-a.low, free) | |||
| } | |||
+ 0
- 83
vendor/github.com/rabbitmq/amqp091-go/auth.go
View File
| @ -1,83 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| package amqp091 | |||
| import ( | |||
| "bytes" | |||
| "fmt" | |||
| ) | |||
| // Authentication interface provides a means for different SASL authentication | |||
| // mechanisms to be used during connection tuning. | |||
| type Authentication interface { | |||
| Mechanism() string | |||
| Response() string | |||
| } | |||
| // PlainAuth is a similar to Basic Auth in HTTP. | |||
| type PlainAuth struct { | |||
| Username string | |||
| Password string | |||
| } | |||
| // Mechanism returns "PLAIN" | |||
| func (auth *PlainAuth) Mechanism() string { | |||
| return "PLAIN" | |||
| } | |||
| // Response returns the null character delimited encoding for the SASL PLAIN Mechanism. | |||
| func (auth *PlainAuth) Response() string { | |||
| return fmt.Sprintf("\000%s\000%s", auth.Username, auth.Password) | |||
| } | |||
| // AMQPlainAuth is similar to PlainAuth | |||
| type AMQPlainAuth struct { | |||
| Username string | |||
| Password string | |||
| } | |||
| // Mechanism returns "AMQPLAIN" | |||
| func (auth *AMQPlainAuth) Mechanism() string { | |||
| return "AMQPLAIN" | |||
| } | |||
| // Response returns an AMQP encoded credentials table, without the field table size. | |||
| func (auth *AMQPlainAuth) Response() string { | |||
| var buf bytes.Buffer | |||
| table := Table{"LOGIN": auth.Username, "PASSWORD": auth.Password} | |||
| if err := writeTable(&buf, table); err != nil { | |||
| return "" | |||
| } | |||
| return buf.String()[4:] | |||
| } | |||
| // ExternalAuth for RabbitMQ-auth-mechanism-ssl. | |||
| type ExternalAuth struct { | |||
| } | |||
| // Mechanism returns "EXTERNAL" | |||
| func (*ExternalAuth) Mechanism() string { | |||
| return "EXTERNAL" | |||
| } | |||
| // Response returns an AMQP encoded credentials table, without the field table size. | |||
| func (*ExternalAuth) Response() string { | |||
| return "\000*\000*" | |||
| } | |||
| // Finds the first mechanism preferred by the client that the server supports. | |||
| func pickSASLMechanism(client []Authentication, serverMechanisms []string) (auth Authentication, ok bool) { | |||
| for _, auth = range client { | |||
| for _, mech := range serverMechanisms { | |||
| if auth.Mechanism() == mech { | |||
| return auth, true | |||
| } | |||
| } | |||
| } | |||
| return | |||
| } | |||
+ 0
- 159
vendor/github.com/rabbitmq/amqp091-go/certs.sh
View File
| @ -1,159 +0,0 @@ | |||
| #!/bin/sh | |||
| # | |||
| # Creates the CA, server and client certs to be used by tls_test.go | |||
| # http://www.rabbitmq.com/ssl.html | |||
| # | |||
| # Copy stdout into the const section of tls_test.go or use for RabbitMQ | |||
| # | |||
| root=$PWD/certs | |||
| if [ -f $root/ca/serial ]; then | |||
| echo >&2 "Previous installation found" | |||
| echo >&2 "Remove $root/ca and rerun to overwrite" | |||
| exit 1 | |||
| fi | |||
| mkdir -p $root/ca/private | |||
| mkdir -p $root/ca/certs | |||
| mkdir -p $root/server | |||
| mkdir -p $root/client | |||
| cd $root/ca | |||
| chmod 700 private | |||
| touch index.txt | |||
| echo 'unique_subject = no' > index.txt.attr | |||
| echo '01' > serial | |||
| echo >openssl.cnf ' | |||
| [ ca ] | |||
| default_ca = testca | |||
| [ testca ] | |||
| dir = . | |||
| certificate = $dir/cacert.pem | |||
| database = $dir/index.txt | |||
| new_certs_dir = $dir/certs | |||
| private_key = $dir/private/cakey.pem | |||
| serial = $dir/serial | |||
| default_crl_days = 7 | |||
| default_days = 3650 | |||
| default_md = sha1 | |||
| policy = testca_policy | |||
| x509_extensions = certificate_extensions | |||
| [ testca_policy ] | |||
| commonName = supplied | |||
| stateOrProvinceName = optional | |||
| countryName = optional | |||
| emailAddress = optional | |||
| organizationName = optional | |||
| organizationalUnitName = optional | |||
| [ certificate_extensions ] | |||
| basicConstraints = CA:false | |||
| [ req ] | |||
| default_bits = 2048 | |||
| default_keyfile = ./private/cakey.pem | |||
| default_md = sha1 | |||
| prompt = yes | |||
| distinguished_name = root_ca_distinguished_name | |||
| x509_extensions = root_ca_extensions | |||
| [ root_ca_distinguished_name ] | |||
| commonName = hostname | |||
| [ root_ca_extensions ] | |||
| basicConstraints = CA:true | |||
| keyUsage = keyCertSign, cRLSign | |||
| [ client_ca_extensions ] | |||
| basicConstraints = CA:false | |||
| keyUsage = digitalSignature | |||
| extendedKeyUsage = 1.3.6.1.5.5.7.3.2 | |||
| [ server_ca_extensions ] | |||
| basicConstraints = CA:false | |||
| keyUsage = keyEncipherment | |||
| extendedKeyUsage = 1.3.6.1.5.5.7.3.1 | |||
| subjectAltName = @alt_names | |||
| [ alt_names ] | |||
| IP.1 = 127.0.0.1 | |||
| ' | |||
| openssl req \ | |||
| -x509 \ | |||
| -nodes \ | |||
| -config openssl.cnf \ | |||
| -newkey rsa:2048 \ | |||
| -days 3650 \ | |||
| -subj "/CN=MyTestCA/" \ | |||
| -out cacert.pem \ | |||
| -outform PEM | |||
| openssl x509 \ | |||
| -in cacert.pem \ | |||
| -out cacert.cer \ | |||
| -outform DER | |||
| openssl genrsa -out $root/server/key.pem 2048 | |||
| openssl genrsa -out $root/client/key.pem 2048 | |||
| openssl req \ | |||
| -new \ | |||
| -nodes \ | |||
| -config openssl.cnf \ | |||
| -subj "/CN=127.0.0.1/O=server/" \ | |||
| -key $root/server/key.pem \ | |||
| -out $root/server/req.pem \ | |||
| -outform PEM | |||
| openssl req \ | |||
| -new \ | |||
| -nodes \ | |||
| -config openssl.cnf \ | |||
| -subj "/CN=127.0.0.1/O=client/" \ | |||
| -key $root/client/key.pem \ | |||
| -out $root/client/req.pem \ | |||
| -outform PEM | |||
| openssl ca \ | |||
| -config openssl.cnf \ | |||
| -in $root/server/req.pem \ | |||
| -out $root/server/cert.pem \ | |||
| -notext \ | |||
| -batch \ | |||
| -extensions server_ca_extensions | |||
| openssl ca \ | |||
| -config openssl.cnf \ | |||
| -in $root/client/req.pem \ | |||
| -out $root/client/cert.pem \ | |||
| -notext \ | |||
| -batch \ | |||
| -extensions client_ca_extensions | |||
| cat <<-END | |||
| const caCert = \` | |||
| `cat $root/ca/cacert.pem` | |||
| \` | |||
| const serverCert = \` | |||
| `cat $root/server/cert.pem` | |||
| \` | |||
| const serverKey = \` | |||
| `cat $root/server/key.pem` | |||
| \` | |||
| const clientCert = \` | |||
| `cat $root/client/cert.pem` | |||
| \` | |||
| const clientKey = \` | |||
| `cat $root/client/key.pem` | |||
| \` | |||
| END | |||
+ 0
- 4
vendor/github.com/rabbitmq/amqp091-go/change_version.sh
View File
| @ -1,4 +0,0 @@ | |||
| #/bin/bash | |||
| echo $1 > VERSION | |||
| sed -i -e "s/.*buildVersion = \"*.*/buildVersion = \"$1\"/" ./connection.go | |||
| go fmt ./... | |||
+ 0
- 1622
vendor/github.com/rabbitmq/amqp091-go/channel.go
File diff suppressed because it is too large
View File
+ 0
- 180
vendor/github.com/rabbitmq/amqp091-go/confirms.go
View File
| @ -1,180 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| package amqp091 | |||
| import ( | |||
| "sync" | |||
| ) | |||
| // confirms resequences and notifies one or multiple publisher confirmation listeners | |||
| type confirms struct { | |||
| m sync.Mutex | |||
| listeners []chan Confirmation | |||
| sequencer map[uint64]Confirmation | |||
| deferredConfirmations *deferredConfirmations | |||
| published uint64 | |||
| publishedMut sync.Mutex | |||
| expecting uint64 | |||
| } | |||
| // newConfirms allocates a confirms | |||
| func newConfirms() *confirms { | |||
| return &confirms{ | |||
| sequencer: map[uint64]Confirmation{}, | |||
| deferredConfirmations: newDeferredConfirmations(), | |||
| published: 0, | |||
| expecting: 1, | |||
| } | |||
| } | |||
| func (c *confirms) Listen(l chan Confirmation) { | |||
| c.m.Lock() | |||
| defer c.m.Unlock() | |||
| c.listeners = append(c.listeners, l) | |||
| } | |||
| // Publish increments the publishing counter | |||
| func (c *confirms) Publish() *DeferredConfirmation { | |||
| c.publishedMut.Lock() | |||
| defer c.publishedMut.Unlock() | |||
| c.published++ | |||
| return c.deferredConfirmations.Add(c.published) | |||
| } | |||
| // confirm confirms one publishing, increments the expecting delivery tag, and | |||
| // removes bookkeeping for that delivery tag. | |||
| func (c *confirms) confirm(confirmation Confirmation) { | |||
| delete(c.sequencer, c.expecting) | |||
| c.expecting++ | |||
| for _, l := range c.listeners { | |||
| l <- confirmation | |||
| } | |||
| } | |||
| // resequence confirms any out of order delivered confirmations | |||
| func (c *confirms) resequence() { | |||
| c.publishedMut.Lock() | |||
| defer c.publishedMut.Unlock() | |||
| for c.expecting <= c.published { | |||
| sequenced, found := c.sequencer[c.expecting] | |||
| if !found { | |||
| return | |||
| } | |||
| c.confirm(sequenced) | |||
| } | |||
| } | |||
| // One confirms one publishing and all following in the publishing sequence | |||
| func (c *confirms) One(confirmed Confirmation) { | |||
| c.m.Lock() | |||
| defer c.m.Unlock() | |||
| c.deferredConfirmations.Confirm(confirmed) | |||
| if c.expecting == confirmed.DeliveryTag { | |||
| c.confirm(confirmed) | |||
| } else { | |||
| c.sequencer[confirmed.DeliveryTag] = confirmed | |||
| } | |||
| c.resequence() | |||
| } | |||
| // Multiple confirms all publishings up until the delivery tag | |||
| func (c *confirms) Multiple(confirmed Confirmation) { | |||
| c.m.Lock() | |||
| defer c.m.Unlock() | |||
| c.deferredConfirmations.ConfirmMultiple(confirmed) | |||
| for c.expecting <= confirmed.DeliveryTag { | |||
| c.confirm(Confirmation{c.expecting, confirmed.Ack}) | |||
| } | |||
| c.resequence() | |||
| } | |||
| // Cleans up the confirms struct and its dependencies. | |||
| // Closes all listeners, discarding any out of sequence confirmations | |||
| func (c *confirms) Close() error { | |||
| c.m.Lock() | |||
| defer c.m.Unlock() | |||
| c.deferredConfirmations.Close() | |||
| for _, l := range c.listeners { | |||
| close(l) | |||
| } | |||
| c.listeners = nil | |||
| return nil | |||
| } | |||
| type deferredConfirmations struct { | |||
| m sync.Mutex | |||
| confirmations map[uint64]*DeferredConfirmation | |||
| } | |||
| func newDeferredConfirmations() *deferredConfirmations { | |||
| return &deferredConfirmations{ | |||
| confirmations: map[uint64]*DeferredConfirmation{}, | |||
| } | |||
| } | |||
| func (d *deferredConfirmations) Add(tag uint64) *DeferredConfirmation { | |||
| d.m.Lock() | |||
| defer d.m.Unlock() | |||
| dc := &DeferredConfirmation{DeliveryTag: tag} | |||
| dc.wg.Add(1) | |||
| d.confirmations[tag] = dc | |||
| return dc | |||
| } | |||
| func (d *deferredConfirmations) Confirm(confirmation Confirmation) { | |||
| d.m.Lock() | |||
| defer d.m.Unlock() | |||
| dc, found := d.confirmations[confirmation.DeliveryTag] | |||
| if !found { | |||
| // we should never receive a confirmation for a tag that hasn't been published, but a test causes this to happen | |||
| return | |||
| } | |||
| dc.confirmation = confirmation | |||
| dc.wg.Done() | |||
| delete(d.confirmations, confirmation.DeliveryTag) | |||
| } | |||
| func (d *deferredConfirmations) ConfirmMultiple(confirmation Confirmation) { | |||
| d.m.Lock() | |||
| defer d.m.Unlock() | |||
| for k, v := range d.confirmations { | |||
| if k <= confirmation.DeliveryTag { | |||
| v.confirmation = Confirmation{DeliveryTag: k, Ack: confirmation.Ack} | |||
| v.wg.Done() | |||
| delete(d.confirmations, k) | |||
| } | |||
| } | |||
| } | |||
| // Nacks all pending DeferredConfirmations being blocked by dc.Wait() | |||
| func (d *deferredConfirmations) Close() { | |||
| d.m.Lock() | |||
| defer d.m.Unlock() | |||
| for k, v := range d.confirmations { | |||
| v.confirmation = Confirmation{DeliveryTag: k, Ack: false} | |||
| v.wg.Done() | |||
| delete(d.confirmations, k) | |||
| } | |||
| } | |||
| // Waits for publisher confirmation. Returns true if server successfully received the publishing. | |||
| func (d *DeferredConfirmation) Wait() bool { | |||
| d.wg.Wait() | |||
| return d.confirmation.Ack | |||
| } | |||
+ 0
- 875
vendor/github.com/rabbitmq/amqp091-go/connection.go
View File
| @ -1,875 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| package amqp091 | |||
| import ( | |||
| "bufio" | |||
| "crypto/tls" | |||
| "io" | |||
| "net" | |||
| "reflect" | |||
| "strconv" | |||
| "strings" | |||
| "sync" | |||
| "sync/atomic" | |||
| "time" | |||
| ) | |||
| const ( | |||
| maxChannelMax = (2 << 15) - 1 | |||
| defaultHeartbeat = 10 * time.Second | |||
| defaultConnectionTimeout = 30 * time.Second | |||
| defaultProduct = "Amqp 0.9.1 Client" | |||
| buildVersion = "1.3.4" | |||
| platform = "golang" | |||
| // Safer default that makes channel leaks a lot easier to spot | |||
| // before they create operational headaches. See https://github.com/rabbitmq/rabbitmq-server/issues/1593. | |||
| defaultChannelMax = (2 << 10) - 1 | |||
| defaultLocale = "en_US" | |||
| ) | |||
| // Config is used in DialConfig and Open to specify the desired tuning | |||
| // parameters used during a connection open handshake. The negotiated tuning | |||
| // will be stored in the returned connection's Config field. | |||
| type Config struct { | |||
| // The SASL mechanisms to try in the client request, and the successful | |||
| // mechanism used on the Connection object. | |||
| // If SASL is nil, PlainAuth from the URL is used. | |||
| SASL []Authentication | |||
| // Vhost specifies the namespace of permissions, exchanges, queues and | |||
| // bindings on the server. Dial sets this to the path parsed from the URL. | |||
| Vhost string | |||
| ChannelMax int // 0 max channels means 2^16 - 1 | |||
| FrameSize int // 0 max bytes means unlimited | |||
| Heartbeat time.Duration // less than 1s uses the server's interval | |||
| // TLSClientConfig specifies the client configuration of the TLS connection | |||
| // when establishing a tls transport. | |||
| // If the URL uses an amqps scheme, then an empty tls.Config with the | |||
| // ServerName from the URL is used. | |||
| TLSClientConfig *tls.Config | |||
| // Properties is table of properties that the client advertises to the server. | |||
| // This is an optional setting - if the application does not set this, | |||
| // the underlying library will use a generic set of client properties. | |||
| Properties Table | |||
| // Connection locale that we expect to always be en_US | |||
| // Even though servers must return it as per the AMQP 0-9-1 spec, | |||
| // we are not aware of it being used other than to satisfy the spec requirements | |||
| Locale string | |||
| // Dial returns a net.Conn prepared for a TLS handshake with TSLClientConfig, | |||
| // then an AMQP connection handshake. | |||
| // If Dial is nil, net.DialTimeout with a 30s connection and 30s deadline is | |||
| // used during TLS and AMQP handshaking. | |||
| Dial func(network, addr string) (net.Conn, error) | |||
| } | |||
| // Connection manages the serialization and deserialization of frames from IO | |||
| // and dispatches the frames to the appropriate channel. All RPC methods and | |||
| // asynchronous Publishing, Delivery, Ack, Nack and Return messages are | |||
| // multiplexed on this channel. There must always be active receivers for | |||
| // every asynchronous message on this connection. | |||
| type Connection struct { | |||
| destructor sync.Once // shutdown once | |||
| sendM sync.Mutex // conn writer mutex | |||
| m sync.Mutex // struct field mutex | |||
| conn io.ReadWriteCloser | |||
| rpc chan message | |||
| writer *writer | |||
| sends chan time.Time // timestamps of each frame sent | |||
| deadlines chan readDeadliner // heartbeater updates read deadlines | |||
| allocator *allocator // id generator valid after openTune | |||
| channels map[uint16]*Channel | |||
| noNotify bool // true when we will never notify again | |||
| closes []chan *Error | |||
| blocks []chan Blocking | |||
| errors chan *Error | |||
| Config Config // The negotiated Config after connection.open | |||
| Major int // Server's major version | |||
| Minor int // Server's minor version | |||
| Properties Table // Server properties | |||
| Locales []string // Server locales | |||
| closed int32 // Will be 1 if the connection is closed, 0 otherwise. Should only be accessed as atomic | |||
| } | |||
| type readDeadliner interface { | |||
| SetReadDeadline(time.Time) error | |||
| } | |||
| // DefaultDial establishes a connection when config.Dial is not provided | |||
| func DefaultDial(connectionTimeout time.Duration) func(network, addr string) (net.Conn, error) { | |||
| return func(network, addr string) (net.Conn, error) { | |||
| conn, err := net.DialTimeout(network, addr, connectionTimeout) | |||
| if err != nil { | |||
| return nil, err | |||
| } | |||
| // Heartbeating hasn't started yet, don't stall forever on a dead server. | |||
| // A deadline is set for TLS and AMQP handshaking. After AMQP is established, | |||
| // the deadline is cleared in openComplete. | |||
| if err := conn.SetDeadline(time.Now().Add(connectionTimeout)); err != nil { | |||
| return nil, err | |||
| } | |||
| return conn, nil | |||
| } | |||
| } | |||
| // Dial accepts a string in the AMQP URI format and returns a new Connection | |||
| // over TCP using PlainAuth. Defaults to a server heartbeat interval of 10 | |||
| // seconds and sets the handshake deadline to 30 seconds. After handshake, | |||
| // deadlines are cleared. | |||
| // | |||
| // Dial uses the zero value of tls.Config when it encounters an amqps:// | |||
| // scheme. It is equivalent to calling DialTLS(amqp, nil). | |||
| func Dial(url string) (*Connection, error) { | |||
| return DialConfig(url, Config{ | |||
| Heartbeat: defaultHeartbeat, | |||
| Locale: defaultLocale, | |||
| }) | |||
| } | |||
| // DialTLS accepts a string in the AMQP URI format and returns a new Connection | |||
| // over TCP using PlainAuth. Defaults to a server heartbeat interval of 10 | |||
| // seconds and sets the initial read deadline to 30 seconds. | |||
| // | |||
| // DialTLS uses the provided tls.Config when encountering an amqps:// scheme. | |||
| func DialTLS(url string, amqps *tls.Config) (*Connection, error) { | |||
| return DialConfig(url, Config{ | |||
| Heartbeat: defaultHeartbeat, | |||
| TLSClientConfig: amqps, | |||
| Locale: defaultLocale, | |||
| }) | |||
| } | |||
| // DialTLS_ExternalAuth accepts a string in the AMQP URI format and returns a | |||
| // new Connection over TCP using EXTERNAL auth. Defaults to a server heartbeat | |||
| // interval of 10 seconds and sets the initial read deadline to 30 seconds. | |||
| // | |||
| // This mechanism is used, when RabbitMQ is configured for EXTERNAL auth with | |||
| // ssl_cert_login plugin for userless/passwordless logons | |||
| // | |||
| // DialTLS_ExternalAuth uses the provided tls.Config when encountering an | |||
| // amqps:// scheme. | |||
| func DialTLS_ExternalAuth(url string, amqps *tls.Config) (*Connection, error) { | |||
| return DialConfig(url, Config{ | |||
| Heartbeat: defaultHeartbeat, | |||
| TLSClientConfig: amqps, | |||
| SASL: []Authentication{&ExternalAuth{}}, | |||
| }) | |||
| } | |||
| // DialConfig accepts a string in the AMQP URI format and a configuration for | |||
| // the transport and connection setup, returning a new Connection. Defaults to | |||
| // a server heartbeat interval of 10 seconds and sets the initial read deadline | |||
| // to 30 seconds. | |||
| func DialConfig(url string, config Config) (*Connection, error) { | |||
| var err error | |||
| var conn net.Conn | |||
| uri, err := ParseURI(url) | |||
| if err != nil { | |||
| return nil, err | |||
| } | |||
| if config.SASL == nil { | |||
| config.SASL = []Authentication{uri.PlainAuth()} | |||
| } | |||
| if config.Vhost == "" { | |||
| config.Vhost = uri.Vhost | |||
| } | |||
| addr := net.JoinHostPort(uri.Host, strconv.FormatInt(int64(uri.Port), 10)) | |||
| dialer := config.Dial | |||
| if dialer == nil { | |||
| dialer = DefaultDial(defaultConnectionTimeout) | |||
| } | |||
| conn, err = dialer("tcp", addr) | |||
| if err != nil { | |||
| return nil, err | |||
| } | |||
| if uri.Scheme == "amqps" { | |||
| if config.TLSClientConfig == nil { | |||
| config.TLSClientConfig = new(tls.Config) | |||
| } | |||
| // If ServerName has not been specified in TLSClientConfig, | |||
| // set it to the URI host used for this connection. | |||
| if config.TLSClientConfig.ServerName == "" { | |||
| config.TLSClientConfig.ServerName = uri.Host | |||
| } | |||
| client := tls.Client(conn, config.TLSClientConfig) | |||
| if err := client.Handshake(); err != nil { | |||
| conn.Close() | |||
| return nil, err | |||
| } | |||
| conn = client | |||
| } | |||
| return Open(conn, config) | |||
| } | |||
| /* | |||
| Open accepts an already established connection, or other io.ReadWriteCloser as | |||
| a transport. Use this method if you have established a TLS connection or wish | |||
| to use your own custom transport. | |||
| */ | |||
| func Open(conn io.ReadWriteCloser, config Config) (*Connection, error) { | |||
| c := &Connection{ | |||
| conn: conn, | |||
| writer: &writer{bufio.NewWriter(conn)}, | |||
| channels: make(map[uint16]*Channel), | |||
| rpc: make(chan message), | |||
| sends: make(chan time.Time), | |||
| errors: make(chan *Error, 1), | |||
| deadlines: make(chan readDeadliner, 1), | |||
| } | |||
| go c.reader(conn) | |||
| return c, c.open(config) | |||
| } | |||
| /* | |||
| LocalAddr returns the local TCP peer address, or ":0" (the zero value of net.TCPAddr) | |||
| as a fallback default value if the underlying transport does not support LocalAddr(). | |||
| */ | |||
| func (c *Connection) LocalAddr() net.Addr { | |||
| if conn, ok := c.conn.(interface { | |||
| LocalAddr() net.Addr | |||
| }); ok { | |||
| return conn.LocalAddr() | |||
| } | |||
| return &net.TCPAddr{} | |||
| } | |||
| // ConnectionState returns basic TLS details of the underlying transport. | |||
| // Returns a zero value when the underlying connection does not implement | |||
| // ConnectionState() tls.ConnectionState. | |||
| func (c *Connection) ConnectionState() tls.ConnectionState { | |||
| if conn, ok := c.conn.(interface { | |||
| ConnectionState() tls.ConnectionState | |||
| }); ok { | |||
| return conn.ConnectionState() | |||
| } | |||
| return tls.ConnectionState{} | |||
| } | |||
| /* | |||
| NotifyClose registers a listener for close events either initiated by an error | |||
| accompanying a connection.close method or by a normal shutdown. | |||
| The chan provided will be closed when the Channel is closed and on a | |||
| graceful close, no error will be sent. | |||
| In case of a non graceful close the error will be notified synchronously by the library | |||
| so that it will be necessary to consume the Channel from the caller in order to avoid deadlocks | |||
| To reconnect after a transport or protocol error, register a listener here and | |||
| re-run your setup process. | |||
| */ | |||
| func (c *Connection) NotifyClose(receiver chan *Error) chan *Error { | |||
| c.m.Lock() | |||
| defer c.m.Unlock() | |||
| if c.noNotify { | |||
| close(receiver) | |||
| } else { | |||
| c.closes = append(c.closes, receiver) | |||
| } | |||
| return receiver | |||
| } | |||
| /* | |||
| NotifyBlocked registers a listener for RabbitMQ specific TCP flow control | |||
| method extensions connection.blocked and connection.unblocked. Flow control is | |||
| active with a reason when Blocking.Blocked is true. When a Connection is | |||
| blocked, all methods will block across all connections until server resources | |||
| become free again. | |||
| This optional extension is supported by the server when the | |||
| "connection.blocked" server capability key is true. | |||
| */ | |||
| func (c *Connection) NotifyBlocked(receiver chan Blocking) chan Blocking { | |||
| c.m.Lock() | |||
| defer c.m.Unlock() | |||
| if c.noNotify { | |||
| close(receiver) | |||
| } else { | |||
| c.blocks = append(c.blocks, receiver) | |||
| } | |||
| return receiver | |||
| } | |||
| /* | |||
| Close requests and waits for the response to close the AMQP connection. | |||
| It's advisable to use this message when publishing to ensure all kernel buffers | |||
| have been flushed on the server and client before exiting. | |||
| An error indicates that server may not have received this request to close but | |||
| the connection should be treated as closed regardless. | |||
| After returning from this call, all resources associated with this connection, | |||
| including the underlying io, Channels, Notify listeners and Channel consumers | |||
| will also be closed. | |||
| */ | |||
| func (c *Connection) Close() error { | |||
| if c.IsClosed() { | |||
| return ErrClosed | |||
| } | |||
| defer c.shutdown(nil) | |||
| return c.call( | |||
| &connectionClose{ | |||
| ReplyCode: replySuccess, | |||
| ReplyText: "kthxbai", | |||
| }, | |||
| &connectionCloseOk{}, | |||
| ) | |||
| } | |||
| func (c *Connection) closeWith(err *Error) error { | |||
| if c.IsClosed() { | |||
| return ErrClosed | |||
| } | |||
| defer c.shutdown(err) | |||
| return c.call( | |||
| &connectionClose{ | |||
| ReplyCode: uint16(err.Code), | |||
| ReplyText: err.Reason, | |||
| }, | |||
| &connectionCloseOk{}, | |||
| ) | |||
| } | |||
| // IsClosed returns true if the connection is marked as closed, otherwise false | |||
| // is returned. | |||
| func (c *Connection) IsClosed() bool { | |||
| return (atomic.LoadInt32(&c.closed) == 1) | |||
| } | |||
| func (c *Connection) send(f frame) error { | |||
| if c.IsClosed() { | |||
| return ErrClosed | |||
| } | |||
| c.sendM.Lock() | |||
| err := c.writer.WriteFrame(f) | |||
| c.sendM.Unlock() | |||
| if err != nil { | |||
| // shutdown could be re-entrant from signaling notify chans | |||
| go c.shutdown(&Error{ | |||
| Code: FrameError, | |||
| Reason: err.Error(), | |||
| }) | |||
| } else { | |||
| // Broadcast we sent a frame, reducing heartbeats, only | |||
| // if there is something that can receive - like a non-reentrant | |||
| // call or if the heartbeater isn't running | |||
| select { | |||
| case c.sends <- time.Now(): | |||
| default: | |||
| } | |||
| } | |||
| return err | |||
| } | |||
| func (c *Connection) shutdown(err *Error) { | |||
| atomic.StoreInt32(&c.closed, 1) | |||
| c.destructor.Do(func() { | |||
| c.m.Lock() | |||
| defer c.m.Unlock() | |||
| if err != nil { | |||
| for _, c := range c.closes { | |||
| c <- err | |||
| } | |||
| } | |||
| if err != nil { | |||
| c.errors <- err | |||
| } | |||
| // Shutdown handler goroutine can still receive the result. | |||
| close(c.errors) | |||
| for _, c := range c.closes { | |||
| close(c) | |||
| } | |||
| for _, c := range c.blocks { | |||
| close(c) | |||
| } | |||
| // Shutdown the channel, but do not use closeChannel() as it calls | |||
| // releaseChannel() which requires the connection lock. | |||
| // | |||
| // Ranging over c.channels and calling releaseChannel() that mutates | |||
| // c.channels is racy - see commit 6063341 for an example. | |||
| for _, ch := range c.channels { | |||
| ch.shutdown(err) | |||
| } | |||
| c.conn.Close() | |||
| c.channels = map[uint16]*Channel{} | |||
| c.allocator = newAllocator(1, c.Config.ChannelMax) | |||
| c.noNotify = true | |||
| }) | |||
| } | |||
| // All methods sent to the connection channel should be synchronous so we | |||
| // can handle them directly without a framing component | |||
| func (c *Connection) demux(f frame) { | |||
| if f.channel() == 0 { | |||
| c.dispatch0(f) | |||
| } else { | |||
| c.dispatchN(f) | |||
| } | |||
| } | |||
| func (c *Connection) dispatch0(f frame) { | |||
| switch mf := f.(type) { | |||
| case *methodFrame: | |||
| switch m := mf.Method.(type) { | |||
| case *connectionClose: | |||
| // Send immediately as shutdown will close our side of the writer. | |||
| c.send(&methodFrame{ | |||
| ChannelId: 0, | |||
| Method: &connectionCloseOk{}, | |||
| }) | |||
| c.shutdown(newError(m.ReplyCode, m.ReplyText)) | |||
| case *connectionBlocked: | |||
| for _, c := range c.blocks { | |||
| c <- Blocking{Active: true, Reason: m.Reason} | |||
| } | |||
| case *connectionUnblocked: | |||
| for _, c := range c.blocks { | |||
| c <- Blocking{Active: false} | |||
| } | |||
| default: | |||
| c.rpc <- m | |||
| } | |||
| case *heartbeatFrame: | |||
| // kthx - all reads reset our deadline. so we can drop this | |||
| default: | |||
| // lolwat - channel0 only responds to methods and heartbeats | |||
| c.closeWith(ErrUnexpectedFrame) | |||
| } | |||
| } | |||
| func (c *Connection) dispatchN(f frame) { | |||
| c.m.Lock() | |||
| channel := c.channels[f.channel()] | |||
| c.m.Unlock() | |||
| if channel != nil { | |||
| channel.recv(channel, f) | |||
| } else { | |||
| c.dispatchClosed(f) | |||
| } | |||
| } | |||
| // section 2.3.7: "When a peer decides to close a channel or connection, it | |||
| // sends a Close method. The receiving peer MUST respond to a Close with a | |||
| // Close-Ok, and then both parties can close their channel or connection. Note | |||
| // that if peers ignore Close, deadlock can happen when both peers send Close | |||
| // at the same time." | |||
| // | |||
| // When we don't have a channel, so we must respond with close-ok on a close | |||
| // method. This can happen between a channel exception on an asynchronous | |||
| // method like basic.publish and a synchronous close with channel.close. | |||
| // In that case, we'll get both a channel.close and channel.close-ok in any | |||
| // order. | |||
| func (c *Connection) dispatchClosed(f frame) { | |||
| // Only consider method frames, drop content/header frames | |||
| if mf, ok := f.(*methodFrame); ok { | |||
| switch mf.Method.(type) { | |||
| case *channelClose: | |||
| c.send(&methodFrame{ | |||
| ChannelId: f.channel(), | |||
| Method: &channelCloseOk{}, | |||
| }) | |||
| case *channelCloseOk: | |||
| // we are already closed, so do nothing | |||
| default: | |||
| // unexpected method on closed channel | |||
| c.closeWith(ErrClosed) | |||
| } | |||
| } | |||
| } | |||
| // Reads each frame off the IO and hand off to the connection object that | |||
| // will demux the streams and dispatch to one of the opened channels or | |||
| // handle on channel 0 (the connection channel). | |||
| func (c *Connection) reader(r io.Reader) { | |||
| buf := bufio.NewReader(r) | |||
| frames := &reader{buf} | |||
| conn, haveDeadliner := r.(readDeadliner) | |||
| for { | |||
| frame, err := frames.ReadFrame() | |||
| if err != nil { | |||
| c.shutdown(&Error{Code: FrameError, Reason: err.Error()}) | |||
| return | |||
| } | |||
| c.demux(frame) | |||
| if haveDeadliner { | |||
| select { | |||
| case c.deadlines <- conn: | |||
| default: | |||
| // On c.Close() c.heartbeater() might exit just before c.deadlines <- conn is called. | |||
| // Which results in this goroutine being stuck forever. | |||
| } | |||
| } | |||
| } | |||
| } | |||
| // Ensures that at least one frame is being sent at the tuned interval with a | |||
| // jitter tolerance of 1s | |||
| func (c *Connection) heartbeater(interval time.Duration, done chan *Error) { | |||
| const maxServerHeartbeatsInFlight = 3 | |||
| var sendTicks <-chan time.Time | |||
| if interval > 0 { | |||
| ticker := time.NewTicker(interval) | |||
| defer ticker.Stop() | |||
| sendTicks = ticker.C | |||
| } | |||
| lastSent := time.Now() | |||
| for { | |||
| select { | |||
| case at, stillSending := <-c.sends: | |||
| // When actively sending, depend on sent frames to reset server timer | |||
| if stillSending { | |||
| lastSent = at | |||
| } else { | |||
| return | |||
| } | |||
| case at := <-sendTicks: | |||
| // When idle, fill the space with a heartbeat frame | |||
| if at.Sub(lastSent) > interval-time.Second { | |||
| if err := c.send(&heartbeatFrame{}); err != nil { | |||
| // send heartbeats even after close/closeOk so we | |||
| // tick until the connection starts erroring | |||
| return | |||
| } | |||
| } | |||
| case conn := <-c.deadlines: | |||
| // When reading, reset our side of the deadline, if we've negotiated one with | |||
| // a deadline that covers at least 2 server heartbeats | |||
| if interval > 0 { | |||
| conn.SetReadDeadline(time.Now().Add(maxServerHeartbeatsInFlight * interval)) | |||
| } | |||
| case <-done: | |||
| return | |||
| } | |||
| } | |||
| } | |||
| // Convenience method to inspect the Connection.Properties["capabilities"] | |||
| // Table for server identified capabilities like "basic.ack" or | |||
| // "confirm.select". | |||
| func (c *Connection) isCapable(featureName string) bool { | |||
| capabilities, _ := c.Properties["capabilities"].(Table) | |||
| hasFeature, _ := capabilities[featureName].(bool) | |||
| return hasFeature | |||
| } | |||
| // allocateChannel records but does not open a new channel with a unique id. | |||
| // This method is the initial part of the channel lifecycle and paired with | |||
| // releaseChannel | |||
| func (c *Connection) allocateChannel() (*Channel, error) { | |||
| c.m.Lock() | |||
| defer c.m.Unlock() | |||
| if c.IsClosed() { | |||
| return nil, ErrClosed | |||
| } | |||
| id, ok := c.allocator.next() | |||
| if !ok { | |||
| return nil, ErrChannelMax | |||
| } | |||
| ch := newChannel(c, uint16(id)) | |||
| c.channels[uint16(id)] = ch | |||
| return ch, nil | |||
| } | |||
| // releaseChannel removes a channel from the registry as the final part of the | |||
| // channel lifecycle | |||
| func (c *Connection) releaseChannel(id uint16) { | |||
| c.m.Lock() | |||
| defer c.m.Unlock() | |||
| delete(c.channels, id) | |||
| c.allocator.release(int(id)) | |||
| } | |||
| // openChannel allocates and opens a channel, must be paired with closeChannel | |||
| func (c *Connection) openChannel() (*Channel, error) { | |||
| ch, err := c.allocateChannel() | |||
| if err != nil { | |||
| return nil, err | |||
| } | |||
| if err := ch.open(); err != nil { | |||
| c.releaseChannel(ch.id) | |||
| return nil, err | |||
| } | |||
| return ch, nil | |||
| } | |||
| // closeChannel releases and initiates a shutdown of the channel. All channel | |||
| // closures should be initiated here for proper channel lifecycle management on | |||
| // this connection. | |||
| func (c *Connection) closeChannel(ch *Channel, e *Error) { | |||
| ch.shutdown(e) | |||
| c.releaseChannel(ch.id) | |||
| } | |||
| /* | |||
| Channel opens a unique, concurrent server channel to process the bulk of AMQP | |||
| messages. Any error from methods on this receiver will render the receiver | |||
| invalid and a new Channel should be opened. | |||
| */ | |||
| func (c *Connection) Channel() (*Channel, error) { | |||
| return c.openChannel() | |||
| } | |||
| func (c *Connection) call(req message, res ...message) error { | |||
| // Special case for when the protocol header frame is sent insted of a | |||
| // request method | |||
| if req != nil { | |||
| if err := c.send(&methodFrame{ChannelId: 0, Method: req}); err != nil { | |||
| return err | |||
| } | |||
| } | |||
| select { | |||
| case err, ok := <-c.errors: | |||
| if !ok { | |||
| return ErrClosed | |||
| } | |||
| return err | |||
| case msg := <-c.rpc: | |||
| // Try to match one of the result types | |||
| for _, try := range res { | |||
| if reflect.TypeOf(msg) == reflect.TypeOf(try) { | |||
| // *res = *msg | |||
| vres := reflect.ValueOf(try).Elem() | |||
| vmsg := reflect.ValueOf(msg).Elem() | |||
| vres.Set(vmsg) | |||
| return nil | |||
| } | |||
| } | |||
| return ErrCommandInvalid | |||
| } | |||
| // unreachable | |||
| } | |||
| // Connection = open-Connection *use-Connection close-Connection | |||
| // open-Connection = C:protocol-header | |||
| // S:START C:START-OK | |||
| // *challenge | |||
| // S:TUNE C:TUNE-OK | |||
| // C:OPEN S:OPEN-OK | |||
| // challenge = S:SECURE C:SECURE-OK | |||
| // use-Connection = *channel | |||
| // close-Connection = C:CLOSE S:CLOSE-OK | |||
| // / S:CLOSE C:CLOSE-OK | |||
| func (c *Connection) open(config Config) error { | |||
| if err := c.send(&protocolHeader{}); err != nil { | |||
| return err | |||
| } | |||
| return c.openStart(config) | |||
| } | |||
| func (c *Connection) openStart(config Config) error { | |||
| start := &connectionStart{} | |||
| if err := c.call(nil, start); err != nil { | |||
| return err | |||
| } | |||
| c.Major = int(start.VersionMajor) | |||
| c.Minor = int(start.VersionMinor) | |||
| c.Properties = start.ServerProperties | |||
| c.Locales = strings.Split(start.Locales, " ") | |||
| // eventually support challenge/response here by also responding to | |||
| // connectionSecure. | |||
| auth, ok := pickSASLMechanism(config.SASL, strings.Split(start.Mechanisms, " ")) | |||
| if !ok { | |||
| return ErrSASL | |||
| } | |||
| // Save this mechanism off as the one we chose | |||
| c.Config.SASL = []Authentication{auth} | |||
| // Set the connection locale to client locale | |||
| c.Config.Locale = config.Locale | |||
| return c.openTune(config, auth) | |||
| } | |||
| func (c *Connection) openTune(config Config, auth Authentication) error { | |||
| if len(config.Properties) == 0 { | |||
| config.Properties = Table{ | |||
| "product": defaultProduct, | |||
| "version": buildVersion, | |||
| "platform": platform, | |||
| } | |||
| } | |||
| config.Properties["capabilities"] = Table{ | |||
| "connection.blocked": true, | |||
| "consumer_cancel_notify": true, | |||
| } | |||
| ok := &connectionStartOk{ | |||
| ClientProperties: config.Properties, | |||
| Mechanism: auth.Mechanism(), | |||
| Response: auth.Response(), | |||
| Locale: config.Locale, | |||
| } | |||
| tune := &connectionTune{} | |||
| if err := c.call(ok, tune); err != nil { | |||
| // per spec, a connection can only be closed when it has been opened | |||
| // so at this point, we know it's an auth error, but the socket | |||
| // was closed instead. Return a meaningful error. | |||
| return ErrCredentials | |||
| } | |||
| // When the server and client both use default 0, then the max channel is | |||
| // only limited by uint16. | |||
| c.Config.ChannelMax = pick(config.ChannelMax, int(tune.ChannelMax)) | |||
| if c.Config.ChannelMax == 0 { | |||
| c.Config.ChannelMax = defaultChannelMax | |||
| } | |||
| c.Config.ChannelMax = min(c.Config.ChannelMax, maxChannelMax) | |||
| // Frame size includes headers and end byte (len(payload)+8), even if | |||
| // this is less than FrameMinSize, use what the server sends because the | |||
| // alternative is to stop the handshake here. | |||
| c.Config.FrameSize = pick(config.FrameSize, int(tune.FrameMax)) | |||
| // Save this off for resetDeadline() | |||
| c.Config.Heartbeat = time.Second * time.Duration(pick( | |||
| int(config.Heartbeat/time.Second), | |||
| int(tune.Heartbeat))) | |||
| // "The client should start sending heartbeats after receiving a | |||
| // Connection.Tune method" | |||
| go c.heartbeater(c.Config.Heartbeat/2, c.NotifyClose(make(chan *Error, 1))) | |||
| if err := c.send(&methodFrame{ | |||
| ChannelId: 0, | |||
| Method: &connectionTuneOk{ | |||
| ChannelMax: uint16(c.Config.ChannelMax), | |||
| FrameMax: uint32(c.Config.FrameSize), | |||
| Heartbeat: uint16(c.Config.Heartbeat / time.Second), | |||
| }, | |||
| }); err != nil { | |||
| return err | |||
| } | |||
| return c.openVhost(config) | |||
| } | |||
| func (c *Connection) openVhost(config Config) error { | |||
| req := &connectionOpen{VirtualHost: config.Vhost} | |||
| res := &connectionOpenOk{} | |||
| if err := c.call(req, res); err != nil { | |||
| // Cannot be closed yet, but we know it's a vhost problem | |||
| return ErrVhost | |||
| } | |||
| c.Config.Vhost = config.Vhost | |||
| return c.openComplete() | |||
| } | |||
| // openComplete performs any final Connection initialization dependent on the | |||
| // connection handshake and clears any state needed for TLS and AMQP handshaking. | |||
| func (c *Connection) openComplete() error { | |||
| // We clear the deadlines and let the heartbeater reset the read deadline if requested. | |||
| // RabbitMQ uses TCP flow control at this point for pushback so Writes can | |||
| // intentionally block. | |||
| if deadliner, ok := c.conn.(interface { | |||
| SetDeadline(time.Time) error | |||
| }); ok { | |||
| _ = deadliner.SetDeadline(time.Time{}) | |||
| } | |||
| c.allocator = newAllocator(1, c.Config.ChannelMax) | |||
| return nil | |||
| } | |||
| func max(a, b int) int { | |||
| if a > b { | |||
| return a | |||
| } | |||
| return b | |||
| } | |||
| func min(a, b int) int { | |||
| if a < b { | |||
| return a | |||
| } | |||
| return b | |||
| } | |||
| func pick(client, server int) int { | |||
| if client == 0 || server == 0 { | |||
| return max(client, server) | |||
| } | |||
| return min(client, server) | |||
| } | |||
+ 0
- 142
vendor/github.com/rabbitmq/amqp091-go/consumers.go
View File
| @ -1,142 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| package amqp091 | |||
| import ( | |||
| "os" | |||
| "strconv" | |||
| "sync" | |||
| "sync/atomic" | |||
| ) | |||
| var consumerSeq uint64 | |||
| const consumerTagLengthMax = 0xFF // see writeShortstr | |||
| func uniqueConsumerTag() string { | |||
| return commandNameBasedUniqueConsumerTag(os.Args[0]) | |||
| } | |||
| func commandNameBasedUniqueConsumerTag(commandName string) string { | |||
| tagPrefix := "ctag-" | |||
| tagInfix := commandName | |||
| tagSuffix := "-" + strconv.FormatUint(atomic.AddUint64(&consumerSeq, 1), 10) | |||
| if len(tagPrefix)+len(tagInfix)+len(tagSuffix) > consumerTagLengthMax { | |||
| tagInfix = "streadway/amqp" | |||
| } | |||
| return tagPrefix + tagInfix + tagSuffix | |||
| } | |||
| type consumerBuffers map[string]chan *Delivery | |||
| // Concurrent type that manages the consumerTag -> | |||
| // ingress consumerBuffer mapping | |||
| type consumers struct { | |||
| sync.WaitGroup // one for buffer | |||
| closed chan struct{} // signal buffer | |||
| sync.Mutex // protects below | |||
| chans consumerBuffers | |||
| } | |||
| func makeConsumers() *consumers { | |||
| return &consumers{ | |||
| closed: make(chan struct{}), | |||
| chans: make(consumerBuffers), | |||
| } | |||
| } | |||
| func (subs *consumers) buffer(in chan *Delivery, out chan Delivery) { | |||
| defer close(out) | |||
| defer subs.Done() | |||
| var inflight = in | |||
| var queue []*Delivery | |||
| for delivery := range in { | |||
| queue = append(queue, delivery) | |||
| for len(queue) > 0 { | |||
| select { | |||
| case <-subs.closed: | |||
| // closed before drained, drop in-flight | |||
| return | |||
| case delivery, consuming := <-inflight: | |||
| if consuming { | |||
| queue = append(queue, delivery) | |||
| } else { | |||
| inflight = nil | |||
| } | |||
| case out <- *queue[0]: | |||
| queue = queue[1:] | |||
| } | |||
| } | |||
| } | |||
| } | |||
| // On key conflict, close the previous channel. | |||
| func (subs *consumers) add(tag string, consumer chan Delivery) { | |||
| subs.Lock() | |||
| defer subs.Unlock() | |||
| if prev, found := subs.chans[tag]; found { | |||
| close(prev) | |||
| } | |||
| in := make(chan *Delivery) | |||
| subs.chans[tag] = in | |||
| subs.Add(1) | |||
| go subs.buffer(in, consumer) | |||
| } | |||
| func (subs *consumers) cancel(tag string) (found bool) { | |||
| subs.Lock() | |||
| defer subs.Unlock() | |||
| ch, found := subs.chans[tag] | |||
| if found { | |||
| delete(subs.chans, tag) | |||
| close(ch) | |||
| } | |||
| return found | |||
| } | |||
| func (subs *consumers) close() { | |||
| subs.Lock() | |||
| defer subs.Unlock() | |||
| close(subs.closed) | |||
| for tag, ch := range subs.chans { | |||
| delete(subs.chans, tag) | |||
| close(ch) | |||
| } | |||
| subs.Wait() | |||
| } | |||
| // Sends a delivery to a the consumer identified by `tag`. | |||
| // If unbuffered channels are used for Consume this method | |||
| // could block all deliveries until the consumer | |||
| // receives on the other end of the channel. | |||
| func (subs *consumers) send(tag string, msg *Delivery) bool { | |||
| subs.Lock() | |||
| defer subs.Unlock() | |||
| buffer, found := subs.chans[tag] | |||
| if found { | |||
| buffer <- msg | |||
| } | |||
| return found | |||
| } | |||
+ 0
- 173
vendor/github.com/rabbitmq/amqp091-go/delivery.go
View File
| @ -1,173 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| package amqp091 | |||
| import ( | |||
| "errors" | |||
| "time" | |||
| ) | |||
| var errDeliveryNotInitialized = errors.New("delivery not initialized") | |||
| // Acknowledger notifies the server of successful or failed consumption of | |||
| // deliveries via identifier found in the Delivery.DeliveryTag field. | |||
| // | |||
| // Applications can provide mock implementations in tests of Delivery handlers. | |||
| type Acknowledger interface { | |||
| Ack(tag uint64, multiple bool) error | |||
| Nack(tag uint64, multiple bool, requeue bool) error | |||
| Reject(tag uint64, requeue bool) error | |||
| } | |||
| // Delivery captures the fields for a previously delivered message resident in | |||
| // a queue to be delivered by the server to a consumer from Channel.Consume or | |||
| // Channel.Get. | |||
| type Delivery struct { | |||
| Acknowledger Acknowledger // the channel from which this delivery arrived | |||
| Headers Table // Application or header exchange table | |||
| // Properties | |||
| ContentType string // MIME content type | |||
| ContentEncoding string // MIME content encoding | |||
| DeliveryMode uint8 // queue implementation use - non-persistent (1) or persistent (2) | |||
| Priority uint8 // queue implementation use - 0 to 9 | |||
| CorrelationId string // application use - correlation identifier | |||
| ReplyTo string // application use - address to reply to (ex: RPC) | |||
| Expiration string // implementation use - message expiration spec | |||
| MessageId string // application use - message identifier | |||
| Timestamp time.Time // application use - message timestamp | |||
| Type string // application use - message type name | |||
| UserId string // application use - creating user - should be authenticated user | |||
| AppId string // application use - creating application id | |||
| // Valid only with Channel.Consume | |||
| ConsumerTag string | |||
| // Valid only with Channel.Get | |||
| MessageCount uint32 | |||
| DeliveryTag uint64 | |||
| Redelivered bool | |||
| Exchange string // basic.publish exchange | |||
| RoutingKey string // basic.publish routing key | |||
| Body []byte | |||
| } | |||
| func newDelivery(channel *Channel, msg messageWithContent) *Delivery { | |||
| props, body := msg.getContent() | |||
| delivery := Delivery{ | |||
| Acknowledger: channel, | |||
| Headers: props.Headers, | |||
| ContentType: props.ContentType, | |||
| ContentEncoding: props.ContentEncoding, | |||
| DeliveryMode: props.DeliveryMode, | |||
| Priority: props.Priority, | |||
| CorrelationId: props.CorrelationId, | |||
| ReplyTo: props.ReplyTo, | |||
| Expiration: props.Expiration, | |||
| MessageId: props.MessageId, | |||
| Timestamp: props.Timestamp, | |||
| Type: props.Type, | |||
| UserId: props.UserId, | |||
| AppId: props.AppId, | |||
| Body: body, | |||
| } | |||
| // Properties for the delivery types | |||
| switch m := msg.(type) { | |||
| case *basicDeliver: | |||
| delivery.ConsumerTag = m.ConsumerTag | |||
| delivery.DeliveryTag = m.DeliveryTag | |||
| delivery.Redelivered = m.Redelivered | |||
| delivery.Exchange = m.Exchange | |||
| delivery.RoutingKey = m.RoutingKey | |||
| case *basicGetOk: | |||
| delivery.MessageCount = m.MessageCount | |||
| delivery.DeliveryTag = m.DeliveryTag | |||
| delivery.Redelivered = m.Redelivered | |||
| delivery.Exchange = m.Exchange | |||
| delivery.RoutingKey = m.RoutingKey | |||
| } | |||
| return &delivery | |||
| } | |||
| /* | |||
| Ack delegates an acknowledgement through the Acknowledger interface that the | |||
| client or server has finished work on a delivery. | |||
| All deliveries in AMQP must be acknowledged. If you called Channel.Consume | |||
| with autoAck true then the server will be automatically ack each message and | |||
| this method should not be called. Otherwise, you must call Delivery.Ack after | |||
| you have successfully processed this delivery. | |||
| When multiple is true, this delivery and all prior unacknowledged deliveries | |||
| on the same channel will be acknowledged. This is useful for batch processing | |||
| of deliveries. | |||
| An error will indicate that the acknowledge could not be delivered to the | |||
| channel it was sent from. | |||
| Either Delivery.Ack, Delivery.Reject or Delivery.Nack must be called for every | |||
| delivery that is not automatically acknowledged. | |||
| */ | |||
| func (d Delivery) Ack(multiple bool) error { | |||
| if d.Acknowledger == nil { | |||
| return errDeliveryNotInitialized | |||
| } | |||
| return d.Acknowledger.Ack(d.DeliveryTag, multiple) | |||
| } | |||
| /* | |||
| Reject delegates a negatively acknowledgement through the Acknowledger interface. | |||
| When requeue is true, queue this message to be delivered to a consumer on a | |||
| different channel. When requeue is false or the server is unable to queue this | |||
| message, it will be dropped. | |||
| If you are batch processing deliveries, and your server supports it, prefer | |||
| Delivery.Nack. | |||
| Either Delivery.Ack, Delivery.Reject or Delivery.Nack must be called for every | |||
| delivery that is not automatically acknowledged. | |||
| */ | |||
| func (d Delivery) Reject(requeue bool) error { | |||
| if d.Acknowledger == nil { | |||
| return errDeliveryNotInitialized | |||
| } | |||
| return d.Acknowledger.Reject(d.DeliveryTag, requeue) | |||
| } | |||
| /* | |||
| Nack negatively acknowledge the delivery of message(s) identified by the | |||
| delivery tag from either the client or server. | |||
| When multiple is true, nack messages up to and including delivered messages up | |||
| until the delivery tag delivered on the same channel. | |||
| When requeue is true, request the server to deliver this message to a different | |||
| consumer. If it is not possible or requeue is false, the message will be | |||
| dropped or delivered to a server configured dead-letter queue. | |||
| This method must not be used to select or requeue messages the client wishes | |||
| not to handle, rather it is to inform the server that the client is incapable | |||
| of handling this message at this time. | |||
| Either Delivery.Ack, Delivery.Reject or Delivery.Nack must be called for every | |||
| delivery that is not automatically acknowledged. | |||
| */ | |||
| func (d Delivery) Nack(multiple, requeue bool) error { | |||
| if d.Acknowledger == nil { | |||
| return errDeliveryNotInitialized | |||
| } | |||
| return d.Acknowledger.Nack(d.DeliveryTag, multiple, requeue) | |||
| } | |||
+ 0
- 146
vendor/github.com/rabbitmq/amqp091-go/doc.go
View File
| @ -1,146 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| /* | |||
| Package amqp091 is an AMQP 0.9.1 client with RabbitMQ extensions | |||
| Understand the AMQP 0.9.1 messaging model by reviewing these links first. Much | |||
| of the terminology in this library directly relates to AMQP concepts. | |||
| Resources | |||
| http://www.rabbitmq.com/tutorials/amqp-concepts.html | |||
| http://www.rabbitmq.com/getstarted.html | |||
| http://www.rabbitmq.com/amqp-0-9-1-reference.html | |||
| Design | |||
| Most other broker clients publish to queues, but in AMQP, clients publish | |||
| Exchanges instead. AMQP is programmable, meaning that both the producers and | |||
| consumers agree on the configuration of the broker, instead of requiring an | |||
| operator or system configuration that declares the logical topology in the | |||
| broker. The routing between producers and consumer queues is via Bindings. | |||
| These bindings form the logical topology of the broker. | |||
| In this library, a message sent from publisher is called a "Publishing" and a | |||
| message received to a consumer is called a "Delivery". The fields of | |||
| Publishings and Deliveries are close but not exact mappings to the underlying | |||
| wire format to maintain stronger types. Many other libraries will combine | |||
| message properties with message headers. In this library, the message well | |||
| known properties are strongly typed fields on the Publishings and Deliveries, | |||
| whereas the user defined headers are in the Headers field. | |||
| The method naming closely matches the protocol's method name with positional | |||
| parameters mapping to named protocol message fields. The motivation here is to | |||
| present a comprehensive view over all possible interactions with the server. | |||
| Generally, methods that map to protocol methods of the "basic" class will be | |||
| elided in this interface, and "select" methods of various channel mode selectors | |||
| will be elided for example Channel.Confirm and Channel.Tx. | |||
| The library is intentionally designed to be synchronous, where responses for | |||
| each protocol message are required to be received in an RPC manner. Some | |||
| methods have a noWait parameter like Channel.QueueDeclare, and some methods are | |||
| asynchronous like Channel.Publish. The error values should still be checked for | |||
| these methods as they will indicate IO failures like when the underlying | |||
| connection closes. | |||
| Asynchronous Events | |||
| Clients of this library may be interested in receiving some of the protocol | |||
| messages other than Deliveries like basic.ack methods while a channel is in | |||
| confirm mode. | |||
| The Notify* methods with Connection and Channel receivers model the pattern of | |||
| asynchronous events like closes due to exceptions, or messages that are sent out | |||
| of band from an RPC call like basic.ack or basic.flow. | |||
| Any asynchronous events, including Deliveries and Publishings must always have | |||
| a receiver until the corresponding chans are closed. Without asynchronous | |||
| receivers, the synchronous methods will block. | |||
| Use Case | |||
| It's important as a client to an AMQP topology to ensure the state of the | |||
| broker matches your expectations. For both publish and consume use cases, | |||
| make sure you declare the queues, exchanges and bindings you expect to exist | |||
| prior to calling Channel.Publish or Channel.Consume. | |||
| // Connections start with amqp.Dial() typically from a command line argument | |||
| // or environment variable. | |||
| connection, err := amqp.Dial(os.Getenv("AMQP_URL")) | |||
| // To cleanly shutdown by flushing kernel buffers, make sure to close and | |||
| // wait for the response. | |||
| defer connection.Close() | |||
| // Most operations happen on a channel. If any error is returned on a | |||
| // channel, the channel will no longer be valid, throw it away and try with | |||
| // a different channel. If you use many channels, it's useful for the | |||
| // server to | |||
| channel, err := connection.Channel() | |||
| // Declare your topology here, if it doesn't exist, it will be created, if | |||
| // it existed already and is not what you expect, then that's considered an | |||
| // error. | |||
| // Use your connection on this topology with either Publish or Consume, or | |||
| // inspect your queues with QueueInspect. It's unwise to mix Publish and | |||
| // Consume to let TCP do its job well. | |||
| SSL/TLS - Secure connections | |||
| When Dial encounters an amqps:// scheme, it will use the zero value of a | |||
| tls.Config. This will only perform server certificate and host verification. | |||
| Use DialTLS when you wish to provide a client certificate (recommended), | |||
| include a private certificate authority's certificate in the cert chain for | |||
| server validity, or run insecure by not verifying the server certificate dial | |||
| your own connection. DialTLS will use the provided tls.Config when it | |||
| encounters an amqps:// scheme and will dial a plain connection when it | |||
| encounters an amqp:// scheme. | |||
| SSL/TLS in RabbitMQ is documented here: http://www.rabbitmq.com/ssl.html | |||
| Best practises for Connections and Channels notifications. | |||
| In order to be notified when a connection or channel gets closed both the structures offer the possibility to register channels using the `notifyClose` function like: | |||
| notifyConnClose := make(chan *amqp.Error) | |||
| conn.NotifyClose(notifyConnClose) | |||
| No errors will be sent in case of a graceful connection close. | |||
| In case of a non-graceful close, because of a network issue of forced disconnection from the UI, the error will be notified synchronously by the library. | |||
| You can see that in the shutdown function of connection and channel (see connection.go and channel.go) | |||
| if err != nil { | |||
| for _, c := range c.closes { | |||
| c <- err | |||
| } | |||
| } | |||
| The error is sent synchronously to the channel so that the flow will wait until the channel will be consumed by the caller. | |||
| To avoid deadlocks it is necessary to consume the messages from the channels. | |||
| This could be done inside a different goroutine with a select listening on the two channels inside a for loop like: | |||
| go func() { | |||
| for notifyConnClose != nil || notifyChanClose != nil { | |||
| select { | |||
| case err, ok := <-notifyConnClose: | |||
| if !(ok) { | |||
| notifyConnClose = nil | |||
| } else { | |||
| fmt.Printf("connection closed, error %s", err) | |||
| } | |||
| case err, ok := <-notifyChanClose: | |||
| if !(ok) { | |||
| notifyChanClose = nil | |||
| } else { | |||
| fmt.Printf("channel closed, error %s", err) | |||
| } | |||
| } | |||
| } | |||
| }() | |||
| */ | |||
| package amqp091 | |||
+ 0
- 23
vendor/github.com/rabbitmq/amqp091-go/fuzz.go
View File
| @ -1,23 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| //go:build gofuzz | |||
| // +build gofuzz | |||
| package amqp091 | |||
| import "bytes" | |||
| func Fuzz(data []byte) int { | |||
| r := reader{bytes.NewReader(data)} | |||
| frame, err := r.ReadFrame() | |||
| if err != nil { | |||
| if frame != nil { | |||
| panic("frame is not nil") | |||
| } | |||
| return 0 | |||
| } | |||
| return 1 | |||
| } | |||
+ 0
- 2
vendor/github.com/rabbitmq/amqp091-go/gen.sh
View File
| @ -1,2 +0,0 @@ | |||
| #!/bin/sh | |||
| go run spec/gen.go < spec/amqp0-9-1.stripped.extended.xml | gofmt > spec091.go | |||
+ 0
- 67
vendor/github.com/rabbitmq/amqp091-go/pre-commit
View File
| @ -1,67 +0,0 @@ | |||
| #!/bin/sh | |||
| LATEST_STABLE_SUPPORTED_GO_VERSION="1.11" | |||
| main() { | |||
| if local_go_version_is_latest_stable | |||
| then | |||
| run_gofmt | |||
| run_golint | |||
| run_govet | |||
| fi | |||
| run_unit_tests | |||
| } | |||
| local_go_version_is_latest_stable() { | |||
| go version | grep -q $LATEST_STABLE_SUPPORTED_GO_VERSION | |||
| } | |||
| log_error() { | |||
| echo "$*" 1>&2 | |||
| } | |||
| run_gofmt() { | |||
| GOFMT_FILES=$(gofmt -l .) | |||
| if [ -n "$GOFMT_FILES" ] | |||
| then | |||
| log_error "gofmt failed for the following files: | |||
| $GOFMT_FILES | |||
| please run 'gofmt -w .' on your changes before committing." | |||
| exit 1 | |||
| fi | |||
| } | |||
| run_golint() { | |||
| GOLINT_ERRORS=$(golint ./... | grep -v "Id should be") | |||
| if [ -n "$GOLINT_ERRORS" ] | |||
| then | |||
| log_error "golint failed for the following reasons: | |||
| $GOLINT_ERRORS | |||
| please run 'golint ./...' on your changes before committing." | |||
| exit 1 | |||
| fi | |||
| } | |||
| run_govet() { | |||
| GOVET_ERRORS=$(go tool vet ./*.go 2>&1) | |||
| if [ -n "$GOVET_ERRORS" ] | |||
| then | |||
| log_error "go vet failed for the following reasons: | |||
| $GOVET_ERRORS | |||
| please run 'go tool vet ./*.go' on your changes before committing." | |||
| exit 1 | |||
| fi | |||
| } | |||
| run_unit_tests() { | |||
| if [ -z "$NOTEST" ] | |||
| then | |||
| log_error 'Running short tests...' | |||
| env AMQP_URL= go test -short | |||
| fi | |||
| } | |||
| main | |||
+ 0
- 464
vendor/github.com/rabbitmq/amqp091-go/read.go
View File
| @ -1,464 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| package amqp091 | |||
| import ( | |||
| "bytes" | |||
| "encoding/binary" | |||
| "errors" | |||
| "io" | |||
| "time" | |||
| ) | |||
| /* | |||
| ReadFrame reads a frame from an input stream and returns an interface that can be cast into | |||
| one of the following: | |||
| methodFrame | |||
| PropertiesFrame | |||
| bodyFrame | |||
| heartbeatFrame | |||
| 2.3.5 frame Details | |||
| All frames consist of a header (7 octets), a payload of arbitrary size, and a | |||
| 'frame-end' octet that detects malformed frames: | |||
| 0 1 3 7 size+7 size+8 | |||
| +------+---------+-------------+ +------------+ +-----------+ | |||
| | type | channel | size | | payload | | frame-end | | |||
| +------+---------+-------------+ +------------+ +-----------+ | |||
| octet short long size octets octet | |||
| To read a frame, we: | |||
| 1. Read the header and check the frame type and channel. | |||
| 2. Depending on the frame type, we read the payload and process it. | |||
| 3. Read the frame end octet. | |||
| In realistic implementations where performance is a concern, we would use | |||
| “read-ahead buffering” or | |||
| “gathering reads” to avoid doing three separate system calls to read a frame. | |||
| */ | |||
| func (r *reader) ReadFrame() (frame frame, err error) { | |||
| var scratch [7]byte | |||
| if _, err = io.ReadFull(r.r, scratch[:7]); err != nil { | |||
| return | |||
| } | |||
| typ := scratch[0] | |||
| channel := binary.BigEndian.Uint16(scratch[1:3]) | |||
| size := binary.BigEndian.Uint32(scratch[3:7]) | |||
| switch typ { | |||
| case frameMethod: | |||
| if frame, err = r.parseMethodFrame(channel, size); err != nil { | |||
| return | |||
| } | |||
| case frameHeader: | |||
| if frame, err = r.parseHeaderFrame(channel, size); err != nil { | |||
| return | |||
| } | |||
| case frameBody: | |||
| if frame, err = r.parseBodyFrame(channel, size); err != nil { | |||
| return nil, err | |||
| } | |||
| case frameHeartbeat: | |||
| if frame, err = r.parseHeartbeatFrame(channel, size); err != nil { | |||
| return | |||
| } | |||
| default: | |||
| return nil, ErrFrame | |||
| } | |||
| if _, err = io.ReadFull(r.r, scratch[:1]); err != nil { | |||
| return nil, err | |||
| } | |||
| if scratch[0] != frameEnd { | |||
| return nil, ErrFrame | |||
| } | |||
| return | |||
| } | |||
| func readShortstr(r io.Reader) (v string, err error) { | |||
| var length uint8 | |||
| if err = binary.Read(r, binary.BigEndian, &length); err != nil { | |||
| return | |||
| } | |||
| bytes := make([]byte, length) | |||
| if _, err = io.ReadFull(r, bytes); err != nil { | |||
| return | |||
| } | |||
| return string(bytes), nil | |||
| } | |||
| func readLongstr(r io.Reader) (v string, err error) { | |||
| var length uint32 | |||
| if err = binary.Read(r, binary.BigEndian, &length); err != nil { | |||
| return | |||
| } | |||
| // slices can't be longer than max int32 value | |||
| if length > (^uint32(0) >> 1) { | |||
| return | |||
| } | |||
| bytes := make([]byte, length) | |||
| if _, err = io.ReadFull(r, bytes); err != nil { | |||
| return | |||
| } | |||
| return string(bytes), nil | |||
| } | |||
| func readDecimal(r io.Reader) (v Decimal, err error) { | |||
| if err = binary.Read(r, binary.BigEndian, &v.Scale); err != nil { | |||
| return | |||
| } | |||
| if err = binary.Read(r, binary.BigEndian, &v.Value); err != nil { | |||
| return | |||
| } | |||
| return | |||
| } | |||
| func readFloat32(r io.Reader) (v float32, err error) { | |||
| if err = binary.Read(r, binary.BigEndian, &v); err != nil { | |||
| return | |||
| } | |||
| return | |||
| } | |||
| func readFloat64(r io.Reader) (v float64, err error) { | |||
| if err = binary.Read(r, binary.BigEndian, &v); err != nil { | |||
| return | |||
| } | |||
| return | |||
| } | |||
| func readTimestamp(r io.Reader) (v time.Time, err error) { | |||
| var sec int64 | |||
| if err = binary.Read(r, binary.BigEndian, &sec); err != nil { | |||
| return | |||
| } | |||
| return time.Unix(sec, 0), nil | |||
| } | |||
| /* | |||
| 'A': []interface{} | |||
| 'D': Decimal | |||
| 'F': Table | |||
| 'I': int32 | |||
| 'S': string | |||
| 'T': time.Time | |||
| 'V': nil | |||
| 'b': int8 | |||
| 'B': byte | |||
| 'd': float64 | |||
| 'f': float32 | |||
| 'l': int64 | |||
| 's': int16 | |||
| 't': bool | |||
| 'x': []byte | |||
| */ | |||
| func readField(r io.Reader) (v interface{}, err error) { | |||
| var typ byte | |||
| if err = binary.Read(r, binary.BigEndian, &typ); err != nil { | |||
| return | |||
| } | |||
| switch typ { | |||
| case 't': | |||
| var value uint8 | |||
| if err = binary.Read(r, binary.BigEndian, &value); err != nil { | |||
| return | |||
| } | |||
| return (value != 0), nil | |||
| case 'B': | |||
| var value [1]byte | |||
| if _, err = io.ReadFull(r, value[0:1]); err != nil { | |||
| return | |||
| } | |||
| return value[0], nil | |||
| case 'b': | |||
| var value int8 | |||
| if err = binary.Read(r, binary.BigEndian, &value); err != nil { | |||
| return | |||
| } | |||
| return value, nil | |||
| case 's': | |||
| var value int16 | |||
| if err = binary.Read(r, binary.BigEndian, &value); err != nil { | |||
| return | |||
| } | |||
| return value, nil | |||
| case 'I': | |||
| var value int32 | |||
| if err = binary.Read(r, binary.BigEndian, &value); err != nil { | |||
| return | |||
| } | |||
| return value, nil | |||
| case 'l': | |||
| var value int64 | |||
| if err = binary.Read(r, binary.BigEndian, &value); err != nil { | |||
| return | |||
| } | |||
| return value, nil | |||
| case 'f': | |||
| var value float32 | |||
| if err = binary.Read(r, binary.BigEndian, &value); err != nil { | |||
| return | |||
| } | |||
| return value, nil | |||
| case 'd': | |||
| var value float64 | |||
| if err = binary.Read(r, binary.BigEndian, &value); err != nil { | |||
| return | |||
| } | |||
| return value, nil | |||
| case 'D': | |||
| return readDecimal(r) | |||
| case 'S': | |||
| return readLongstr(r) | |||
| case 'A': | |||
| return readArray(r) | |||
| case 'T': | |||
| return readTimestamp(r) | |||
| case 'F': | |||
| return readTable(r) | |||
| case 'x': | |||
| var len int32 | |||
| if err = binary.Read(r, binary.BigEndian, &len); err != nil { | |||
| return nil, err | |||
| } | |||
| value := make([]byte, len) | |||
| if _, err = io.ReadFull(r, value); err != nil { | |||
| return nil, err | |||
| } | |||
| return value, err | |||
| case 'V': | |||
| return nil, nil | |||
| } | |||
| return nil, ErrSyntax | |||
| } | |||
| /* | |||
| Field tables are long strings that contain packed name-value pairs. The | |||
| name-value pairs are encoded as short string defining the name, and octet | |||
| defining the values type and then the value itself. The valid field types for | |||
| tables are an extension of the native integer, bit, string, and timestamp | |||
| types, and are shown in the grammar. Multi-octet integer fields are always | |||
| held in network byte order. | |||
| */ | |||
| func readTable(r io.Reader) (table Table, err error) { | |||
| var nested bytes.Buffer | |||
| var str string | |||
| if str, err = readLongstr(r); err != nil { | |||
| return | |||
| } | |||
| nested.Write([]byte(str)) | |||
| table = make(Table) | |||
| for nested.Len() > 0 { | |||
| var key string | |||
| var value interface{} | |||
| if key, err = readShortstr(&nested); err != nil { | |||
| return | |||
| } | |||
| if value, err = readField(&nested); err != nil { | |||
| return | |||
| } | |||
| table[key] = value | |||
| } | |||
| return | |||
| } | |||
| func readArray(r io.Reader) ([]interface{}, error) { | |||
| var ( | |||
| size uint32 | |||
| err error | |||
| ) | |||
| if err = binary.Read(r, binary.BigEndian, &size); err != nil { | |||
| return nil, err | |||
| } | |||
| var ( | |||
| lim = &io.LimitedReader{R: r, N: int64(size)} | |||
| arr []interface{} | |||
| field interface{} | |||
| ) | |||
| for { | |||
| if field, err = readField(lim); err != nil { | |||
| if err == io.EOF { | |||
| break | |||
| } | |||
| return nil, err | |||
| } | |||
| arr = append(arr, field) | |||
| } | |||
| return arr, nil | |||
| } | |||
| // Checks if this bit mask matches the flags bitset | |||
| func hasProperty(mask uint16, prop int) bool { | |||
| return int(mask)&prop > 0 | |||
| } | |||
| func (r *reader) parseHeaderFrame(channel uint16, size uint32) (frame frame, err error) { | |||
| hf := &headerFrame{ | |||
| ChannelId: channel, | |||
| } | |||
| if err = binary.Read(r.r, binary.BigEndian, &hf.ClassId); err != nil { | |||
| return | |||
| } | |||
| if err = binary.Read(r.r, binary.BigEndian, &hf.weight); err != nil { | |||
| return | |||
| } | |||
| if err = binary.Read(r.r, binary.BigEndian, &hf.Size); err != nil { | |||
| return | |||
| } | |||
| var flags uint16 | |||
| if err = binary.Read(r.r, binary.BigEndian, &flags); err != nil { | |||
| return | |||
| } | |||
| if hasProperty(flags, flagContentType) { | |||
| if hf.Properties.ContentType, err = readShortstr(r.r); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagContentEncoding) { | |||
| if hf.Properties.ContentEncoding, err = readShortstr(r.r); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagHeaders) { | |||
| if hf.Properties.Headers, err = readTable(r.r); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagDeliveryMode) { | |||
| if err = binary.Read(r.r, binary.BigEndian, &hf.Properties.DeliveryMode); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagPriority) { | |||
| if err = binary.Read(r.r, binary.BigEndian, &hf.Properties.Priority); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagCorrelationId) { | |||
| if hf.Properties.CorrelationId, err = readShortstr(r.r); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagReplyTo) { | |||
| if hf.Properties.ReplyTo, err = readShortstr(r.r); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagExpiration) { | |||
| if hf.Properties.Expiration, err = readShortstr(r.r); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagMessageId) { | |||
| if hf.Properties.MessageId, err = readShortstr(r.r); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagTimestamp) { | |||
| if hf.Properties.Timestamp, err = readTimestamp(r.r); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagType) { | |||
| if hf.Properties.Type, err = readShortstr(r.r); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagUserId) { | |||
| if hf.Properties.UserId, err = readShortstr(r.r); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagAppId) { | |||
| if hf.Properties.AppId, err = readShortstr(r.r); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(flags, flagReserved1) { | |||
| if hf.Properties.reserved1, err = readShortstr(r.r); err != nil { | |||
| return | |||
| } | |||
| } | |||
| return hf, nil | |||
| } | |||
| func (r *reader) parseBodyFrame(channel uint16, size uint32) (frame frame, err error) { | |||
| bf := &bodyFrame{ | |||
| ChannelId: channel, | |||
| Body: make([]byte, size), | |||
| } | |||
| if _, err = io.ReadFull(r.r, bf.Body); err != nil { | |||
| return nil, err | |||
| } | |||
| return bf, nil | |||
| } | |||
| var errHeartbeatPayload = errors.New("Heartbeats should not have a payload") | |||
| func (r *reader) parseHeartbeatFrame(channel uint16, size uint32) (frame frame, err error) { | |||
| hf := &heartbeatFrame{ | |||
| ChannelId: channel, | |||
| } | |||
| if size > 0 { | |||
| return nil, errHeartbeatPayload | |||
| } | |||
| return hf, nil | |||
| } | |||
+ 0
- 64
vendor/github.com/rabbitmq/amqp091-go/return.go
View File
| @ -1,64 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| package amqp091 | |||
| import ( | |||
| "time" | |||
| ) | |||
| // Return captures a flattened struct of fields returned by the server when a | |||
| // Publishing is unable to be delivered either due to the `mandatory` flag set | |||
| // and no route found, or `immediate` flag set and no free consumer. | |||
| type Return struct { | |||
| ReplyCode uint16 // reason | |||
| ReplyText string // description | |||
| Exchange string // basic.publish exchange | |||
| RoutingKey string // basic.publish routing key | |||
| // Properties | |||
| ContentType string // MIME content type | |||
| ContentEncoding string // MIME content encoding | |||
| Headers Table // Application or header exchange table | |||
| DeliveryMode uint8 // queue implementation use - non-persistent (1) or persistent (2) | |||
| Priority uint8 // queue implementation use - 0 to 9 | |||
| CorrelationId string // application use - correlation identifier | |||
| ReplyTo string // application use - address to to reply to (ex: RPC) | |||
| Expiration string // implementation use - message expiration spec | |||
| MessageId string // application use - message identifier | |||
| Timestamp time.Time // application use - message timestamp | |||
| Type string // application use - message type name | |||
| UserId string // application use - creating user id | |||
| AppId string // application use - creating application | |||
| Body []byte | |||
| } | |||
| func newReturn(msg basicReturn) *Return { | |||
| props, body := msg.getContent() | |||
| return &Return{ | |||
| ReplyCode: msg.ReplyCode, | |||
| ReplyText: msg.ReplyText, | |||
| Exchange: msg.Exchange, | |||
| RoutingKey: msg.RoutingKey, | |||
| Headers: props.Headers, | |||
| ContentType: props.ContentType, | |||
| ContentEncoding: props.ContentEncoding, | |||
| DeliveryMode: props.DeliveryMode, | |||
| Priority: props.Priority, | |||
| CorrelationId: props.CorrelationId, | |||
| ReplyTo: props.ReplyTo, | |||
| Expiration: props.Expiration, | |||
| MessageId: props.MessageId, | |||
| Timestamp: props.Timestamp, | |||
| Type: props.Type, | |||
| UserId: props.UserId, | |||
| AppId: props.AppId, | |||
| Body: body, | |||
| } | |||
| } | |||
+ 0
- 3306
vendor/github.com/rabbitmq/amqp091-go/spec091.go
File diff suppressed because it is too large
View File
+ 0
- 439
vendor/github.com/rabbitmq/amqp091-go/types.go
View File
| @ -1,439 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| package amqp091 | |||
| import ( | |||
| "fmt" | |||
| "io" | |||
| "sync" | |||
| "time" | |||
| ) | |||
| // Constants for standard AMQP 0-9-1 exchange types. | |||
| const ( | |||
| ExchangeDirect = "direct" | |||
| ExchangeFanout = "fanout" | |||
| ExchangeTopic = "topic" | |||
| ExchangeHeaders = "headers" | |||
| ) | |||
| var ( | |||
| // ErrClosed is returned when the channel or connection is not open | |||
| ErrClosed = &Error{Code: ChannelError, Reason: "channel/connection is not open"} | |||
| // ErrChannelMax is returned when Connection.Channel has been called enough | |||
| // times that all channel IDs have been exhausted in the client or the | |||
| // server. | |||
| ErrChannelMax = &Error{Code: ChannelError, Reason: "channel id space exhausted"} | |||
| // ErrSASL is returned from Dial when the authentication mechanism could not | |||
| // be negotiated. | |||
| ErrSASL = &Error{Code: AccessRefused, Reason: "SASL could not negotiate a shared mechanism"} | |||
| // ErrCredentials is returned when the authenticated client is not authorized | |||
| // to any vhost. | |||
| ErrCredentials = &Error{Code: AccessRefused, Reason: "username or password not allowed"} | |||
| // ErrVhost is returned when the authenticated user is not permitted to | |||
| // access the requested Vhost. | |||
| ErrVhost = &Error{Code: AccessRefused, Reason: "no access to this vhost"} | |||
| // ErrSyntax is hard protocol error, indicating an unsupported protocol, | |||
| // implementation or encoding. | |||
| ErrSyntax = &Error{Code: SyntaxError, Reason: "invalid field or value inside of a frame"} | |||
| // ErrFrame is returned when the protocol frame cannot be read from the | |||
| // server, indicating an unsupported protocol or unsupported frame type. | |||
| ErrFrame = &Error{Code: FrameError, Reason: "frame could not be parsed"} | |||
| // ErrCommandInvalid is returned when the server sends an unexpected response | |||
| // to this requested message type. This indicates a bug in this client. | |||
| ErrCommandInvalid = &Error{Code: CommandInvalid, Reason: "unexpected command received"} | |||
| // ErrUnexpectedFrame is returned when something other than a method or | |||
| // heartbeat frame is delivered to the Connection, indicating a bug in the | |||
| // client. | |||
| ErrUnexpectedFrame = &Error{Code: UnexpectedFrame, Reason: "unexpected frame received"} | |||
| // ErrFieldType is returned when writing a message containing a Go type unsupported by AMQP. | |||
| ErrFieldType = &Error{Code: SyntaxError, Reason: "unsupported table field type"} | |||
| ) | |||
| // Error captures the code and reason a channel or connection has been closed | |||
| // by the server. | |||
| type Error struct { | |||
| Code int // constant code from the specification | |||
| Reason string // description of the error | |||
| Server bool // true when initiated from the server, false when from this library | |||
| Recover bool // true when this error can be recovered by retrying later or with different parameters | |||
| } | |||
| func newError(code uint16, text string) *Error { | |||
| return &Error{ | |||
| Code: int(code), | |||
| Reason: text, | |||
| Recover: isSoftExceptionCode(int(code)), | |||
| Server: true, | |||
| } | |||
| } | |||
| func (e Error) Error() string { | |||
| return fmt.Sprintf("Exception (%d) Reason: %q", e.Code, e.Reason) | |||
| } | |||
| // Used by header frames to capture routing and header information | |||
| type properties struct { | |||
| ContentType string // MIME content type | |||
| ContentEncoding string // MIME content encoding | |||
| Headers Table // Application or header exchange table | |||
| DeliveryMode uint8 // queue implementation use - Transient (1) or Persistent (2) | |||
| Priority uint8 // queue implementation use - 0 to 9 | |||
| CorrelationId string // application use - correlation identifier | |||
| ReplyTo string // application use - address to to reply to (ex: RPC) | |||
| Expiration string // implementation use - message expiration spec | |||
| MessageId string // application use - message identifier | |||
| Timestamp time.Time // application use - message timestamp | |||
| Type string // application use - message type name | |||
| UserId string // application use - creating user id | |||
| AppId string // application use - creating application | |||
| reserved1 string // was cluster-id - process for buffer consumption | |||
| } | |||
| // DeliveryMode. Transient means higher throughput but messages will not be | |||
| // restored on broker restart. The delivery mode of publishings is unrelated | |||
| // to the durability of the queues they reside on. Transient messages will | |||
| // not be restored to durable queues, persistent messages will be restored to | |||
| // durable queues and lost on non-durable queues during server restart. | |||
| // | |||
| // This remains typed as uint8 to match Publishing.DeliveryMode. Other | |||
| // delivery modes specific to custom queue implementations are not enumerated | |||
| // here. | |||
| const ( | |||
| Transient uint8 = 1 | |||
| Persistent uint8 = 2 | |||
| ) | |||
| // The property flags are an array of bits that indicate the presence or | |||
| // absence of each property value in sequence. The bits are ordered from most | |||
| // high to low - bit 15 indicates the first property. | |||
| const ( | |||
| flagContentType = 0x8000 | |||
| flagContentEncoding = 0x4000 | |||
| flagHeaders = 0x2000 | |||
| flagDeliveryMode = 0x1000 | |||
| flagPriority = 0x0800 | |||
| flagCorrelationId = 0x0400 | |||
| flagReplyTo = 0x0200 | |||
| flagExpiration = 0x0100 | |||
| flagMessageId = 0x0080 | |||
| flagTimestamp = 0x0040 | |||
| flagType = 0x0020 | |||
| flagUserId = 0x0010 | |||
| flagAppId = 0x0008 | |||
| flagReserved1 = 0x0004 | |||
| ) | |||
| // Queue captures the current server state of the queue on the server returned | |||
| // from Channel.QueueDeclare or Channel.QueueInspect. | |||
| type Queue struct { | |||
| Name string // server confirmed or generated name | |||
| Messages int // count of messages not awaiting acknowledgment | |||
| Consumers int // number of consumers receiving deliveries | |||
| } | |||
| // Publishing captures the client message sent to the server. The fields | |||
| // outside of the Headers table included in this struct mirror the underlying | |||
| // fields in the content frame. They use native types for convenience and | |||
| // efficiency. | |||
| type Publishing struct { | |||
| // Application or exchange specific fields, | |||
| // the headers exchange will inspect this field. | |||
| Headers Table | |||
| // Properties | |||
| ContentType string // MIME content type | |||
| ContentEncoding string // MIME content encoding | |||
| DeliveryMode uint8 // Transient (0 or 1) or Persistent (2) | |||
| Priority uint8 // 0 to 9 | |||
| CorrelationId string // correlation identifier | |||
| ReplyTo string // address to to reply to (ex: RPC) | |||
| Expiration string // message expiration spec | |||
| MessageId string // message identifier | |||
| Timestamp time.Time // message timestamp | |||
| Type string // message type name | |||
| UserId string // creating user id - ex: "guest" | |||
| AppId string // creating application id | |||
| // The application specific payload of the message | |||
| Body []byte | |||
| } | |||
| // Blocking notifies the server's TCP flow control of the Connection. When a | |||
| // server hits a memory or disk alarm it will block all connections until the | |||
| // resources are reclaimed. Use NotifyBlock on the Connection to receive these | |||
| // events. | |||
| type Blocking struct { | |||
| Active bool // TCP pushback active/inactive on server | |||
| Reason string // Server reason for activation | |||
| } | |||
| // DeferredConfirmation represents a future publisher confirm for a message. It | |||
| // allows users to directly correlate a publishing to a confirmation. These are | |||
| // returned from PublishWithDeferredConfirm on Channels. | |||
| type DeferredConfirmation struct { | |||
| wg sync.WaitGroup | |||
| DeliveryTag uint64 | |||
| confirmation Confirmation | |||
| } | |||
| // Confirmation notifies the acknowledgment or negative acknowledgement of a | |||
| // publishing identified by its delivery tag. Use NotifyPublish on the Channel | |||
| // to consume these events. | |||
| type Confirmation struct { | |||
| DeliveryTag uint64 // A 1 based counter of publishings from when the channel was put in Confirm mode | |||
| Ack bool // True when the server successfully received the publishing | |||
| } | |||
| // Decimal matches the AMQP decimal type. Scale is the number of decimal | |||
| // digits Scale == 2, Value == 12345, Decimal == 123.45 | |||
| type Decimal struct { | |||
| Scale uint8 | |||
| Value int32 | |||
| } | |||
| // Table stores user supplied fields of the following types: | |||
| // | |||
| // bool | |||
| // byte | |||
| // int8 | |||
| // float32 | |||
| // float64 | |||
| // int | |||
| // int16 | |||
| // int32 | |||
| // int64 | |||
| // nil | |||
| // string | |||
| // time.Time | |||
| // amqp.Decimal | |||
| // amqp.Table | |||
| // []byte | |||
| // []interface{} - containing above types | |||
| // | |||
| // Functions taking a table will immediately fail when the table contains a | |||
| // value of an unsupported type. | |||
| // | |||
| // The caller must be specific in which precision of integer it wishes to | |||
| // encode. | |||
| // | |||
| // Use a type assertion when reading values from a table for type conversion. | |||
| // | |||
| // RabbitMQ expects int32 for integer values. | |||
| // | |||
| type Table map[string]interface{} | |||
| func validateField(f interface{}) error { | |||
| switch fv := f.(type) { | |||
| case nil, bool, byte, int8, int, int16, int32, int64, float32, float64, string, []byte, Decimal, time.Time: | |||
| return nil | |||
| case []interface{}: | |||
| for _, v := range fv { | |||
| if err := validateField(v); err != nil { | |||
| return fmt.Errorf("in array %s", err) | |||
| } | |||
| } | |||
| return nil | |||
| case Table: | |||
| for k, v := range fv { | |||
| if err := validateField(v); err != nil { | |||
| return fmt.Errorf("table field %q %s", k, err) | |||
| } | |||
| } | |||
| return nil | |||
| } | |||
| return fmt.Errorf("value %T not supported", f) | |||
| } | |||
| // Validate returns and error if any Go types in the table are incompatible with AMQP types. | |||
| func (t Table) Validate() error { | |||
| return validateField(t) | |||
| } | |||
| // Heap interface for maintaining delivery tags | |||
| type tagSet []uint64 | |||
| func (set tagSet) Len() int { return len(set) } | |||
| func (set tagSet) Less(i, j int) bool { return (set)[i] < (set)[j] } | |||
| func (set tagSet) Swap(i, j int) { (set)[i], (set)[j] = (set)[j], (set)[i] } | |||
| func (set *tagSet) Push(tag interface{}) { *set = append(*set, tag.(uint64)) } | |||
| func (set *tagSet) Pop() interface{} { | |||
| val := (*set)[len(*set)-1] | |||
| *set = (*set)[:len(*set)-1] | |||
| return val | |||
| } | |||
| type message interface { | |||
| id() (uint16, uint16) | |||
| wait() bool | |||
| read(io.Reader) error | |||
| write(io.Writer) error | |||
| } | |||
| type messageWithContent interface { | |||
| message | |||
| getContent() (properties, []byte) | |||
| setContent(properties, []byte) | |||
| } | |||
| /* | |||
| The base interface implemented as: | |||
| 2.3.5 frame Details | |||
| All frames consist of a header (7 octets), a payload of arbitrary size, and a 'frame-end' octet that detects | |||
| malformed frames: | |||
| 0 1 3 7 size+7 size+8 | |||
| +------+---------+-------------+ +------------+ +-----------+ | |||
| | type | channel | size | | payload | | frame-end | | |||
| +------+---------+-------------+ +------------+ +-----------+ | |||
| octet short long size octets octet | |||
| To read a frame, we: | |||
| 1. Read the header and check the frame type and channel. | |||
| 2. Depending on the frame type, we read the payload and process it. | |||
| 3. Read the frame end octet. | |||
| In realistic implementations where performance is a concern, we would use | |||
| “read-ahead buffering” or “gathering reads” to avoid doing three separate | |||
| system calls to read a frame. | |||
| */ | |||
| type frame interface { | |||
| write(io.Writer) error | |||
| channel() uint16 | |||
| } | |||
| type reader struct { | |||
| r io.Reader | |||
| } | |||
| type writer struct { | |||
| w io.Writer | |||
| } | |||
| // Implements the frame interface for Connection RPC | |||
| type protocolHeader struct{} | |||
| func (protocolHeader) write(w io.Writer) error { | |||
| _, err := w.Write([]byte{'A', 'M', 'Q', 'P', 0, 0, 9, 1}) | |||
| return err | |||
| } | |||
| func (protocolHeader) channel() uint16 { | |||
| panic("only valid as initial handshake") | |||
| } | |||
| /* | |||
| Method frames carry the high-level protocol commands (which we call "methods"). | |||
| One method frame carries one command. The method frame payload has this format: | |||
| 0 2 4 | |||
| +----------+-----------+-------------- - - | |||
| | class-id | method-id | arguments... | |||
| +----------+-----------+-------------- - - | |||
| short short ... | |||
| To process a method frame, we: | |||
| 1. Read the method frame payload. | |||
| 2. Unpack it into a structure. A given method always has the same structure, | |||
| so we can unpack the method rapidly. 3. Check that the method is allowed in | |||
| the current context. | |||
| 4. Check that the method arguments are valid. | |||
| 5. Execute the method. | |||
| Method frame bodies are constructed as a list of AMQP data fields (bits, | |||
| integers, strings and string tables). The marshalling code is trivially | |||
| generated directly from the protocol specifications, and can be very rapid. | |||
| */ | |||
| type methodFrame struct { | |||
| ChannelId uint16 | |||
| ClassId uint16 | |||
| MethodId uint16 | |||
| Method message | |||
| } | |||
| func (f *methodFrame) channel() uint16 { return f.ChannelId } | |||
| /* | |||
| Heartbeating is a technique designed to undo one of TCP/IP's features, namely | |||
| its ability to recover from a broken physical connection by closing only after | |||
| a quite long time-out. In some scenarios we need to know very rapidly if a | |||
| peer is disconnected or not responding for other reasons (e.g. it is looping). | |||
| Since heartbeating can be done at a low level, we implement this as a special | |||
| type of frame that peers exchange at the transport level, rather than as a | |||
| class method. | |||
| */ | |||
| type heartbeatFrame struct { | |||
| ChannelId uint16 | |||
| } | |||
| func (f *heartbeatFrame) channel() uint16 { return f.ChannelId } | |||
| /* | |||
| Certain methods (such as Basic.Publish, Basic.Deliver, etc.) are formally | |||
| defined as carrying content. When a peer sends such a method frame, it always | |||
| follows it with a content header and zero or more content body frames. | |||
| A content header frame has this format: | |||
| 0 2 4 12 14 | |||
| +----------+--------+-----------+----------------+------------- - - | |||
| | class-id | weight | body size | property flags | property list... | |||
| +----------+--------+-----------+----------------+------------- - - | |||
| short short long long short remainder... | |||
| We place content body in distinct frames (rather than including it in the | |||
| method) so that AMQP may support "zero copy" techniques in which content is | |||
| never marshalled or encoded. We place the content properties in their own | |||
| frame so that recipients can selectively discard contents they do not want to | |||
| process | |||
| */ | |||
| type headerFrame struct { | |||
| ChannelId uint16 | |||
| ClassId uint16 | |||
| weight uint16 | |||
| Size uint64 | |||
| Properties properties | |||
| } | |||
| func (f *headerFrame) channel() uint16 { return f.ChannelId } | |||
| /* | |||
| Content is the application data we carry from client-to-client via the AMQP | |||
| server. Content is, roughly speaking, a set of properties plus a binary data | |||
| part. The set of allowed properties are defined by the Basic class, and these | |||
| form the "content header frame". The data can be any size, and MAY be broken | |||
| into several (or many) chunks, each forming a "content body frame". | |||
| Looking at the frames for a specific channel, as they pass on the wire, we | |||
| might see something like this: | |||
| [method] | |||
| [method] [header] [body] [body] | |||
| [method] | |||
| ... | |||
| */ | |||
| type bodyFrame struct { | |||
| ChannelId uint16 | |||
| Body []byte | |||
| } | |||
| func (f *bodyFrame) channel() uint16 { return f.ChannelId } | |||
+ 0
- 176
vendor/github.com/rabbitmq/amqp091-go/uri.go
View File
| @ -1,176 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| package amqp091 | |||
| import ( | |||
| "errors" | |||
| "net" | |||
| "net/url" | |||
| "strconv" | |||
| "strings" | |||
| ) | |||
| var errURIScheme = errors.New("AMQP scheme must be either 'amqp://' or 'amqps://'") | |||
| var errURIWhitespace = errors.New("URI must not contain whitespace") | |||
| var schemePorts = map[string]int{ | |||
| "amqp": 5672, | |||
| "amqps": 5671, | |||
| } | |||
| var defaultURI = URI{ | |||
| Scheme: "amqp", | |||
| Host: "localhost", | |||
| Port: 5672, | |||
| Username: "guest", | |||
| Password: "guest", | |||
| Vhost: "/", | |||
| } | |||
| // URI represents a parsed AMQP URI string. | |||
| type URI struct { | |||
| Scheme string | |||
| Host string | |||
| Port int | |||
| Username string | |||
| Password string | |||
| Vhost string | |||
| } | |||
| // ParseURI attempts to parse the given AMQP URI according to the spec. | |||
| // See http://www.rabbitmq.com/uri-spec.html. | |||
| // | |||
| // Default values for the fields are: | |||
| // | |||
| // Scheme: amqp | |||
| // Host: localhost | |||
| // Port: 5672 | |||
| // Username: guest | |||
| // Password: guest | |||
| // Vhost: / | |||
| // | |||
| func ParseURI(uri string) (URI, error) { | |||
| builder := defaultURI | |||
| if strings.Contains(uri, " ") == true { | |||
| return builder, errURIWhitespace | |||
| } | |||
| u, err := url.Parse(uri) | |||
| if err != nil { | |||
| return builder, err | |||
| } | |||
| defaultPort, okScheme := schemePorts[u.Scheme] | |||
| if okScheme { | |||
| builder.Scheme = u.Scheme | |||
| } else { | |||
| return builder, errURIScheme | |||
| } | |||
| host := u.Hostname() | |||
| port := u.Port() | |||
| if host != "" { | |||
| builder.Host = host | |||
| } | |||
| if port != "" { | |||
| port32, err := strconv.ParseInt(port, 10, 32) | |||
| if err != nil { | |||
| return builder, err | |||
| } | |||
| builder.Port = int(port32) | |||
| } else { | |||
| builder.Port = defaultPort | |||
| } | |||
| if u.User != nil { | |||
| builder.Username = u.User.Username() | |||
| if password, ok := u.User.Password(); ok { | |||
| builder.Password = password | |||
| } | |||
| } | |||
| if u.Path != "" { | |||
| if strings.HasPrefix(u.Path, "/") { | |||
| if u.Host == "" && strings.HasPrefix(u.Path, "///") { | |||
| // net/url doesn't handle local context authorities and leaves that up | |||
| // to the scheme handler. In our case, we translate amqp:/// into the | |||
| // default host and whatever the vhost should be | |||
| if len(u.Path) > 3 { | |||
| builder.Vhost = u.Path[3:] | |||
| } | |||
| } else if len(u.Path) > 1 { | |||
| builder.Vhost = u.Path[1:] | |||
| } | |||
| } else { | |||
| builder.Vhost = u.Path | |||
| } | |||
| } | |||
| return builder, nil | |||
| } | |||
| // PlainAuth returns a PlainAuth structure based on the parsed URI's | |||
| // Username and Password fields. | |||
| func (uri URI) PlainAuth() *PlainAuth { | |||
| return &PlainAuth{ | |||
| Username: uri.Username, | |||
| Password: uri.Password, | |||
| } | |||
| } | |||
| // AMQPlainAuth returns a PlainAuth structure based on the parsed URI's | |||
| // Username and Password fields. | |||
| func (uri URI) AMQPlainAuth() *AMQPlainAuth { | |||
| return &AMQPlainAuth{ | |||
| Username: uri.Username, | |||
| Password: uri.Password, | |||
| } | |||
| } | |||
| func (uri URI) String() string { | |||
| authority, err := url.Parse("") | |||
| if err != nil { | |||
| return err.Error() | |||
| } | |||
| authority.Scheme = uri.Scheme | |||
| if uri.Username != defaultURI.Username || uri.Password != defaultURI.Password { | |||
| authority.User = url.User(uri.Username) | |||
| if uri.Password != defaultURI.Password { | |||
| authority.User = url.UserPassword(uri.Username, uri.Password) | |||
| } | |||
| } | |||
| authority.Host = net.JoinHostPort(uri.Host, strconv.Itoa(uri.Port)) | |||
| if defaultPort, found := schemePorts[uri.Scheme]; !found || defaultPort != uri.Port { | |||
| authority.Host = net.JoinHostPort(uri.Host, strconv.Itoa(uri.Port)) | |||
| } else { | |||
| // JoinHostPort() automatically add brackets to the host if it's | |||
| // an IPv6 address. | |||
| // | |||
| // If not port is specified, JoinHostPort() return an IP address in the | |||
| // form of "[::1]:", so we use TrimSuffix() to remove the extra ":". | |||
| authority.Host = strings.TrimSuffix(net.JoinHostPort(uri.Host, ""), ":") | |||
| } | |||
| if uri.Vhost != defaultURI.Vhost { | |||
| // Make sure net/url does not double escape, e.g. | |||
| // "%2F" does not become "%252F". | |||
| authority.Path = uri.Vhost | |||
| authority.RawPath = url.QueryEscape(uri.Vhost) | |||
| } else { | |||
| authority.Path = "/" | |||
| } | |||
| return authority.String() | |||
| } | |||
+ 0
- 422
vendor/github.com/rabbitmq/amqp091-go/write.go
View File
| @ -1,422 +0,0 @@ | |||
| // Copyright (c) 2021 VMware, Inc. or its affiliates. All Rights Reserved. | |||
| // Copyright (c) 2012-2021, Sean Treadway, SoundCloud Ltd. | |||
| // Use of this source code is governed by a BSD-style | |||
| // license that can be found in the LICENSE file. | |||
| package amqp091 | |||
| import ( | |||
| "bufio" | |||
| "bytes" | |||
| "encoding/binary" | |||
| "errors" | |||
| "io" | |||
| "math" | |||
| "time" | |||
| ) | |||
| func (w *writer) WriteFrame(frame frame) (err error) { | |||
| if err = frame.write(w.w); err != nil { | |||
| return | |||
| } | |||
| if buf, ok := w.w.(*bufio.Writer); ok { | |||
| err = buf.Flush() | |||
| } | |||
| return | |||
| } | |||
| func (f *methodFrame) write(w io.Writer) (err error) { | |||
| var payload bytes.Buffer | |||
| if f.Method == nil { | |||
| return errors.New("malformed frame: missing method") | |||
| } | |||
| class, method := f.Method.id() | |||
| if err = binary.Write(&payload, binary.BigEndian, class); err != nil { | |||
| return | |||
| } | |||
| if err = binary.Write(&payload, binary.BigEndian, method); err != nil { | |||
| return | |||
| } | |||
| if err = f.Method.write(&payload); err != nil { | |||
| return | |||
| } | |||
| return writeFrame(w, frameMethod, f.ChannelId, payload.Bytes()) | |||
| } | |||
| // Heartbeat | |||
| // | |||
| // Payload is empty | |||
| func (f *heartbeatFrame) write(w io.Writer) (err error) { | |||
| return writeFrame(w, frameHeartbeat, f.ChannelId, []byte{}) | |||
| } | |||
| // CONTENT HEADER | |||
| // 0 2 4 12 14 | |||
| // +----------+--------+-----------+----------------+------------- - - | |||
| // | class-id | weight | body size | property flags | property list... | |||
| // +----------+--------+-----------+----------------+------------- - - | |||
| // short short long long short remainder... | |||
| // | |||
| func (f *headerFrame) write(w io.Writer) (err error) { | |||
| var payload bytes.Buffer | |||
| var zeroTime time.Time | |||
| if err = binary.Write(&payload, binary.BigEndian, f.ClassId); err != nil { | |||
| return | |||
| } | |||
| if err = binary.Write(&payload, binary.BigEndian, f.weight); err != nil { | |||
| return | |||
| } | |||
| if err = binary.Write(&payload, binary.BigEndian, f.Size); err != nil { | |||
| return | |||
| } | |||
| // First pass will build the mask to be serialized, second pass will serialize | |||
| // each of the fields that appear in the mask. | |||
| var mask uint16 | |||
| if len(f.Properties.ContentType) > 0 { | |||
| mask = mask | flagContentType | |||
| } | |||
| if len(f.Properties.ContentEncoding) > 0 { | |||
| mask = mask | flagContentEncoding | |||
| } | |||
| if f.Properties.Headers != nil && len(f.Properties.Headers) > 0 { | |||
| mask = mask | flagHeaders | |||
| } | |||
| if f.Properties.DeliveryMode > 0 { | |||
| mask = mask | flagDeliveryMode | |||
| } | |||
| if f.Properties.Priority > 0 { | |||
| mask = mask | flagPriority | |||
| } | |||
| if len(f.Properties.CorrelationId) > 0 { | |||
| mask = mask | flagCorrelationId | |||
| } | |||
| if len(f.Properties.ReplyTo) > 0 { | |||
| mask = mask | flagReplyTo | |||
| } | |||
| if len(f.Properties.Expiration) > 0 { | |||
| mask = mask | flagExpiration | |||
| } | |||
| if len(f.Properties.MessageId) > 0 { | |||
| mask = mask | flagMessageId | |||
| } | |||
| if f.Properties.Timestamp != zeroTime { | |||
| mask = mask | flagTimestamp | |||
| } | |||
| if len(f.Properties.Type) > 0 { | |||
| mask = mask | flagType | |||
| } | |||
| if len(f.Properties.UserId) > 0 { | |||
| mask = mask | flagUserId | |||
| } | |||
| if len(f.Properties.AppId) > 0 { | |||
| mask = mask | flagAppId | |||
| } | |||
| if err = binary.Write(&payload, binary.BigEndian, mask); err != nil { | |||
| return | |||
| } | |||
| if hasProperty(mask, flagContentType) { | |||
| if err = writeShortstr(&payload, f.Properties.ContentType); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(mask, flagContentEncoding) { | |||
| if err = writeShortstr(&payload, f.Properties.ContentEncoding); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(mask, flagHeaders) { | |||
| if err = writeTable(&payload, f.Properties.Headers); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(mask, flagDeliveryMode) { | |||
| if err = binary.Write(&payload, binary.BigEndian, f.Properties.DeliveryMode); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(mask, flagPriority) { | |||
| if err = binary.Write(&payload, binary.BigEndian, f.Properties.Priority); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(mask, flagCorrelationId) { | |||
| if err = writeShortstr(&payload, f.Properties.CorrelationId); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(mask, flagReplyTo) { | |||
| if err = writeShortstr(&payload, f.Properties.ReplyTo); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(mask, flagExpiration) { | |||
| if err = writeShortstr(&payload, f.Properties.Expiration); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(mask, flagMessageId) { | |||
| if err = writeShortstr(&payload, f.Properties.MessageId); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(mask, flagTimestamp) { | |||
| if err = binary.Write(&payload, binary.BigEndian, uint64(f.Properties.Timestamp.Unix())); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(mask, flagType) { | |||
| if err = writeShortstr(&payload, f.Properties.Type); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(mask, flagUserId) { | |||
| if err = writeShortstr(&payload, f.Properties.UserId); err != nil { | |||
| return | |||
| } | |||
| } | |||
| if hasProperty(mask, flagAppId) { | |||
| if err = writeShortstr(&payload, f.Properties.AppId); err != nil { | |||
| return | |||
| } | |||
| } | |||
| return writeFrame(w, frameHeader, f.ChannelId, payload.Bytes()) | |||
| } | |||
| // Body | |||
| // | |||
| // Payload is one byterange from the full body who's size is declared in the | |||
| // Header frame | |||
| func (f *bodyFrame) write(w io.Writer) (err error) { | |||
| return writeFrame(w, frameBody, f.ChannelId, f.Body) | |||
| } | |||
| func writeFrame(w io.Writer, typ uint8, channel uint16, payload []byte) (err error) { | |||
| end := []byte{frameEnd} | |||
| size := uint(len(payload)) | |||
| _, err = w.Write([]byte{ | |||
| typ, | |||
| byte((channel & 0xff00) >> 8), | |||
| byte((channel & 0x00ff) >> 0), | |||
| byte((size & 0xff000000) >> 24), | |||
| byte((size & 0x00ff0000) >> 16), | |||
| byte((size & 0x0000ff00) >> 8), | |||
| byte((size & 0x000000ff) >> 0), | |||
| }) | |||
| if err != nil { | |||
| return | |||
| } | |||
| if _, err = w.Write(payload); err != nil { | |||
| return | |||
| } | |||
| if _, err = w.Write(end); err != nil { | |||
| return | |||
| } | |||
| return | |||
| } | |||
| func writeShortstr(w io.Writer, s string) (err error) { | |||
| b := []byte(s) | |||
| var length = uint8(len(b)) | |||
| if err = binary.Write(w, binary.BigEndian, length); err != nil { | |||
| return | |||
| } | |||
| if _, err = w.Write(b[:length]); err != nil { | |||
| return | |||
| } | |||
| return | |||
| } | |||
| func writeLongstr(w io.Writer, s string) (err error) { | |||
| b := []byte(s) | |||
| var length = uint32(len(b)) | |||
| if err = binary.Write(w, binary.BigEndian, length); err != nil { | |||
| return | |||
| } | |||
| if _, err = w.Write(b[:length]); err != nil { | |||
| return | |||
| } | |||
| return | |||
| } | |||
| /* | |||
| 'A': []interface{} | |||
| 'D': Decimal | |||
| 'F': Table | |||
| 'I': int32 | |||
| 'S': string | |||
| 'T': time.Time | |||
| 'V': nil | |||
| 'b': int8 | |||
| 'B': byte | |||
| 'd': float64 | |||
| 'f': float32 | |||
| 'l': int64 | |||
| 's': int16 | |||
| 't': bool | |||
| 'x': []byte | |||
| */ | |||
| func writeField(w io.Writer, value interface{}) (err error) { | |||
| var buf [9]byte | |||
| var enc []byte | |||
| switch v := value.(type) { | |||
| case bool: | |||
| buf[0] = 't' | |||
| if v { | |||
| buf[1] = byte(1) | |||
| } else { | |||
| buf[1] = byte(0) | |||
| } | |||
| enc = buf[:2] | |||
| case byte: | |||
| buf[0] = 'B' | |||
| buf[1] = v | |||
| enc = buf[:2] | |||
| case int8: | |||
| buf[0] = 'b' | |||
| buf[1] = uint8(v) | |||
| enc = buf[:2] | |||
| case int16: | |||
| buf[0] = 's' | |||
| binary.BigEndian.PutUint16(buf[1:3], uint16(v)) | |||
| enc = buf[:3] | |||
| case int: | |||
| buf[0] = 'I' | |||
| binary.BigEndian.PutUint32(buf[1:5], uint32(v)) | |||
| enc = buf[:5] | |||
| case int32: | |||
| buf[0] = 'I' | |||
| binary.BigEndian.PutUint32(buf[1:5], uint32(v)) | |||
| enc = buf[:5] | |||
| case int64: | |||
| buf[0] = 'l' | |||
| binary.BigEndian.PutUint64(buf[1:9], uint64(v)) | |||
| enc = buf[:9] | |||
| case float32: | |||
| buf[0] = 'f' | |||
| binary.BigEndian.PutUint32(buf[1:5], math.Float32bits(v)) | |||
| enc = buf[:5] | |||
| case float64: | |||
| buf[0] = 'd' | |||
| binary.BigEndian.PutUint64(buf[1:9], math.Float64bits(v)) | |||
| enc = buf[:9] | |||
| case Decimal: | |||
| buf[0] = 'D' | |||
| buf[1] = v.Scale | |||
| binary.BigEndian.PutUint32(buf[2:6], uint32(v.Value)) | |||
| enc = buf[:6] | |||
| case string: | |||
| buf[0] = 'S' | |||
| binary.BigEndian.PutUint32(buf[1:5], uint32(len(v))) | |||
| enc = append(buf[:5], []byte(v)...) | |||
| case []interface{}: // field-array | |||
| buf[0] = 'A' | |||
| sec := new(bytes.Buffer) | |||
| for _, val := range v { | |||
| if err = writeField(sec, val); err != nil { | |||
| return | |||
| } | |||
| } | |||
| binary.BigEndian.PutUint32(buf[1:5], uint32(sec.Len())) | |||
| if _, err = w.Write(buf[:5]); err != nil { | |||
| return | |||
| } | |||
| if _, err = w.Write(sec.Bytes()); err != nil { | |||
| return | |||
| } | |||
| return | |||
| case time.Time: | |||
| buf[0] = 'T' | |||
| binary.BigEndian.PutUint64(buf[1:9], uint64(v.Unix())) | |||
| enc = buf[:9] | |||
| case Table: | |||
| if _, err = w.Write([]byte{'F'}); err != nil { | |||
| return | |||
| } | |||
| return writeTable(w, v) | |||
| case []byte: | |||
| buf[0] = 'x' | |||
| binary.BigEndian.PutUint32(buf[1:5], uint32(len(v))) | |||
| if _, err = w.Write(buf[0:5]); err != nil { | |||
| return | |||
| } | |||
| if _, err = w.Write(v); err != nil { | |||
| return | |||
| } | |||
| return | |||
| case nil: | |||
| buf[0] = 'V' | |||
| enc = buf[:1] | |||
| default: | |||
| return ErrFieldType | |||
| } | |||
| _, err = w.Write(enc) | |||
| return | |||
| } | |||
| func writeTable(w io.Writer, table Table) (err error) { | |||
| var buf bytes.Buffer | |||
| for key, val := range table { | |||
| if err = writeShortstr(&buf, key); err != nil { | |||
| return | |||
| } | |||
| if err = writeField(&buf, val); err != nil { | |||
| return | |||
| } | |||
| } | |||
| return writeLongstr(w, string(buf.Bytes())) | |||
| } | |||
+ 0
- 3
vendor/modules.txt
View File
| @ -1,3 +0,0 @@ | |||
| # github.com/rabbitmq/amqp091-go v1.3.4 | |||
| ## explicit; go 1.15 | |||
| github.com/rabbitmq/amqp091-go | |||