Tank/amqp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
Татьяна Фарбер 2025-12-06 13:20:29 +04:00
parent 9a97e05010
commit 53325ac5e2
43 changed files with 10393 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

254
LICENSE
View File

@ -1,117 +1,201 @@
GNU GENERAL PUBLIC LICENSE
Version 2, June 1991
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
Copyright (C) 1989, 1991 Free Software Foundation, Inc.
51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
Everyone is permitted to copy and distribute verbatim copies of this license document, but changing it is not allowed.
1. Definitions.
Preamble
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
The licenses for most software are designed to take away your freedom to share and change it. By contrast, the GNU General Public License is intended to guarantee your freedom to share and change free software--to make sure the software is free for all its users. This General Public License applies to most of the Free Software Foundation's software and to any other program whose authors commit to using it. (Some other Free Software Foundation software is covered by the GNU Lesser General Public License instead.) You can apply it to your programs, too.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
When we speak of free software, we are referring to freedom, not price. Our General Public Licenses are designed to make sure that you have the freedom to distribute copies of free software (and charge for this service if you wish), that you receive source code or can get it if you want it, that you can change the software or use pieces of it in new free programs; and that you know you can do these things.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
To protect your rights, we need to make restrictions that forbid anyone to deny you these rights or to ask you to surrender the rights. These restrictions translate to certain responsibilities for you if you distribute copies of the software, or if you modify it.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
For example, if you distribute copies of such a program, whether gratis or for a fee, you must give the recipients all the rights that you have. You must make sure that they, too, receive or can get the source code. And you must show them these terms so they know their rights.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
We protect your rights with two steps: (1) copyright the software, and (2) offer you this license which gives you legal permission to copy, distribute and/or modify the software.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
Also, for each author's protection and ours, we want to make certain that everyone understands that there is no warranty for this free software. If the software is modified by someone else and passed on, we want its recipients to know that what they have is not the original, so that any problems introduced by others will not reflect on the original authors' reputations.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
Finally, any free program is threatened constantly by software patents. We wish to avoid the danger that redistributors of a free program will individually obtain patent licenses, in effect making the program proprietary. To prevent this, we have made it clear that any patent must be licensed for everyone's free use or not licensed at all.
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
The precise terms and conditions for copying, distribution and modification follow.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
0. This License applies to any program or other work which contains a notice placed by the copyright holder saying it may be distributed under the terms of this General Public License. The "Program", below, refers to any such program or work, and a "work based on the Program" means either the Program or any derivative work under copyright law: that is to say, a work containing the Program or a portion of it, either verbatim or with modifications and/or translated into another language. (Hereinafter, translation is included without limitation in the term "modification".) Each licensee is addressed as "you".
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
Activities other than copying, distribution and modification are not covered by this License; they are outside its scope. The act of running the Program is not restricted, and the output from the Program is covered only if its contents constitute a work based on the Program (independent of having been made by running the Program). Whether that is true depends on what the Program does.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
1. You may copy and distribute verbatim copies of the Program's source code as you receive it, in any medium, provided that you conspicuously and appropriately publish on each copy an appropriate copyright notice and disclaimer of warranty; keep intact all the notices that refer to this License and to the absence of any warranty; and give any other recipients of the Program a copy of this License along with the Program.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
You may charge a fee for the physical act of transferring a copy, and you may at your option offer warranty protection in exchange for a fee.
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
2. You may modify your copy or copies of the Program or any portion of it, thus forming a work based on the Program, and copy and distribute such modifications or work under the terms of Section 1 above, provided that you also meet all of these conditions:
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
a) You must cause the modified files to carry prominent notices stating that you changed the files and the date of any change.
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
b) You must cause any work that you distribute or publish, that in whole or in part contains or is derived from the Program or any part thereof, to be licensed as a whole at no charge to all third parties under the terms of this License.
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
c) If the modified program normally reads commands interactively when run, you must cause it, when started running for such interactive use in the most ordinary way, to print or display an announcement including an appropriate copyright notice and a notice that there is no warranty (or else, saying that you provide a warranty) and that users may redistribute the program under these conditions, and telling the user how to view a copy of this License. (Exception: if the Program itself is interactive but does not normally print such an announcement, your work based on the Program is not required to print an announcement.)
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
These requirements apply to the modified work as a whole. If identifiable sections of that work are not derived from the Program, and can be reasonably considered independent and separate works in themselves, then this License, and its terms, do not apply to those sections when you distribute them as separate works. But when you distribute the same sections as part of a whole which is a work based on the Program, the distribution of the whole must be on the terms of this License, whose permissions for other licensees extend to the entire whole, and thus to each and every part regardless of who wrote it.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
Thus, it is not the intent of this section to claim rights or contest your rights to work written entirely by you; rather, the intent is to exercise the right to control the distribution of derivative or collective works based on the Program.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
In addition, mere aggregation of another work not based on the Program with the Program (or with a work based on the Program) on a volume of a storage or distribution medium does not bring the other work under the scope of this License.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
3. You may copy and distribute the Program (or a work based on it, under Section 2) in object code or executable form under the terms of Sections 1 and 2 above provided that you also do one of the following:
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
a) Accompany it with the complete corresponding machine-readable source code, which must be distributed under the terms of Sections 1 and 2 above on a medium customarily used for software interchange; or,
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
b) Accompany it with a written offer, valid for at least three years, to give any third party, for a charge no more than your cost of physically performing source distribution, a complete machine-readable copy of the corresponding source code, to be distributed under the terms of Sections 1 and 2 above on a medium customarily used for software interchange; or,
END OF TERMS AND CONDITIONS
c) Accompany it with the information you received as to the offer to distribute corresponding source code. (This alternative is allowed only for noncommercial distribution and only if you received the program in object code or executable form with such an offer, in accord with Subsection b above.)
APPENDIX: How to apply the Apache License to your work.
The source code for a work means the preferred form of the work for making modifications to it. For an executable work, complete source code means all the source code for all modules it contains, plus any associated interface definition files, plus the scripts used to control compilation and installation of the executable. However, as a special exception, the source code distributed need not include anything that is normally distributed (in either source or binary form) with the major components (compiler, kernel, and so on) of the operating system on which the executable runs, unless that component itself accompanies the executable.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "{}"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
If distribution of executable or object code is made by offering access to copy from a designated place, then offering equivalent access to copy the source code from the same place counts as distribution of the source code, even though third parties are not compelled to copy the source along with the object code.
Copyright {yyyy} {name of copyright owner}
4. You may not copy, modify, sublicense, or distribute the Program except as expressly provided under this License. Any attempt otherwise to copy, modify, sublicense or distribute the Program is void, and will automatically terminate your rights under this License. However, parties who have received copies, or rights, from you under this License will not have their licenses terminated so long as such parties remain in full compliance.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
5. You are not required to accept this License, since you have not signed it. However, nothing else grants you permission to modify or distribute the Program or its derivative works. These actions are prohibited by law if you do not accept this License. Therefore, by modifying or distributing the Program (or any work based on the Program), you indicate your acceptance of this License to do so, and all its terms and conditions for copying, distributing or modifying the Program or works based on it.
http://www.apache.org/licenses/LICENSE-2.0
6. Each time you redistribute the Program (or any work based on the Program), the recipient automatically receives a license from the original licensor to copy, distribute or modify the Program subject to these terms and conditions. You may not impose any further restrictions on the recipients' exercise of the rights granted herein. You are not responsible for enforcing compliance by third parties to this License.
7. If, as a consequence of a court judgment or allegation of patent infringement or for any other reason (not limited to patent issues), conditions are imposed on you (whether by court order, agreement or otherwise) that contradict the conditions of this License, they do not excuse you from the conditions of this License. If you cannot distribute so as to satisfy simultaneously your obligations under this License and any other pertinent obligations, then as a consequence you may not distribute the Program at all. For example, if a patent license would not permit royalty-free redistribution of the Program by all those who receive copies directly or indirectly through you, then the only way you could satisfy both it and this License would be to refrain entirely from distribution of the Program.
If any portion of this section is held invalid or unenforceable under any particular circumstance, the balance of the section is intended to apply and the section as a whole is intended to apply in other circumstances.
It is not the purpose of this section to induce you to infringe any patents or other property right claims or to contest validity of any such claims; this section has the sole purpose of protecting the integrity of the free software distribution system, which is implemented by public license practices. Many people have made generous contributions to the wide range of software distributed through that system in reliance on consistent application of that system; it is up to the author/donor to decide if he or she is willing to distribute software through any other system and a licensee cannot impose that choice.
This section is intended to make thoroughly clear what is believed to be a consequence of the rest of this License.
8. If the distribution and/or use of the Program is restricted in certain countries either by patents or by copyrighted interfaces, the original copyright holder who places the Program under this License may add an explicit geographical distribution limitation excluding those countries, so that distribution is permitted only in or among countries not thus excluded. In such case, this License incorporates the limitation as if written in the body of this License.
9. The Free Software Foundation may publish revised and/or new versions of the General Public License from time to time. Such new versions will be similar in spirit to the present version, but may differ in detail to address new problems or concerns.
Each version is given a distinguishing version number. If the Program specifies a version number of this License which applies to it and "any later version", you have the option of following the terms and conditions either of that version or of any later version published by the Free Software Foundation. If the Program does not specify a version number of this License, you may choose any version ever published by the Free Software Foundation.
10. If you wish to incorporate parts of the Program into other free programs whose distribution conditions are different, write to the author to ask for permission. For software which is copyrighted by the Free Software Foundation, write to the Free Software Foundation; we sometimes make exceptions for this. Our decision will be guided by the two goals of preserving the free status of all derivatives of our free software and of promoting the sharing and reuse of software generally.
NO WARRANTY
11. BECAUSE THE PROGRAM IS LICENSED FREE OF CHARGE, THERE IS NO WARRANTY FOR THE PROGRAM, TO THE EXTENT PERMITTED BY APPLICABLE LAW. EXCEPT WHEN OTHERWISE STATED IN WRITING THE COPYRIGHT HOLDERS AND/OR OTHER PARTIES PROVIDE THE PROGRAM "AS IS" WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE ENTIRE RISK AS TO THE QUALITY AND PERFORMANCE OF THE PROGRAM IS WITH YOU. SHOULD THE PROGRAM PROVE DEFECTIVE, YOU ASSUME THE COST OF ALL NECESSARY SERVICING, REPAIR OR CORRECTION.
12. IN NO EVENT UNLESS REQUIRED BY APPLICABLE LAW OR AGREED TO IN WRITING WILL ANY COPYRIGHT HOLDER, OR ANY OTHER PARTY WHO MAY MODIFY AND/OR REDISTRIBUTE THE PROGRAM AS PERMITTED ABOVE, BE LIABLE TO YOU FOR DAMAGES, INCLUDING ANY GENERAL, SPECIAL, INCIDENTAL OR CONSEQUENTIAL DAMAGES ARISING OUT OF THE USE OR INABILITY TO USE THE PROGRAM (INCLUDING BUT NOT LIMITED TO LOSS OF DATA OR DATA BEING RENDERED INACCURATE OR LOSSES SUSTAINED BY YOU OR THIRD PARTIES OR A FAILURE OF THE PROGRAM TO OPERATE WITH ANY OTHER PROGRAMS), EVEN IF SUCH HOLDER OR OTHER PARTY HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGES.
END OF TERMS AND CONDITIONS
How to Apply These Terms to Your New Programs
If you develop a new program, and you want it to be of the greatest possible use to the public, the best way to achieve this is to make it free software which everyone can redistribute and change under these terms.
To do so, attach the following notices to the program. It is safest to attach them to the start of each source file to most effectively convey the exclusion of warranty; and each file should have at least the "copyright" line and a pointer to where the full notice is found.
one line to give the program's name and an idea of what it does. Copyright (C) yyyy name of author
This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version.
This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
You should have received a copy of the GNU General Public License along with this program; if not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Also add information on how to contact you by electronic and paper mail.
If the program is interactive, make it output a short notice like this when it starts in an interactive mode:
Gnomovision version 69, Copyright (C) year name of author Gnomovision comes with ABSOLUTELY NO WARRANTY; for details type `show w'. This is free software, and you are welcome to redistribute it under certain conditions; type `show c' for details.
The hypothetical commands `show w' and `show c' should show the appropriate parts of the General Public License. Of course, the commands you use may be called something other than `show w' and `show c'; they could even be mouse-clicks or menu items--whatever suits your program.
You should also get your employer (if you work as a programmer) or your school, if any, to sign a "copyright disclaimer" for the program, if necessary. Here is a sample; alter the names:
Yoyodyne, Inc., hereby disclaims all copyright interest in the program `Gnomovision' (which makes passes at compilers) written by James Hacker.
signature of Ty Coon, 1 April 1989 Ty Coon, President of Vice
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

292
README.md
View File

@ -1,3 +1,291 @@
# amqp
# Lua AMQP Client (support RabbitMQ)
Клиент для AMQP протокола (RabbitMQ и т.д) с поддержкой бесконечного loop
It is a fork of :
https://github.com/mengz0/amqp
https://github.com/ZigzagAK/amqp
https://github.com/4mig4/lua-amqp
https://github.com/gsdenys/amqp-client
This library can be used with LuaJIT and does not have to be used only in OpenResty.
This library use lua socket and ***don't use cqueues***.
***Added reject, requeue, multiple, queue arguments and remove cqueues connects:***
If requeue option sets in false then acknowledgement method sets to NOACK (with multiple option).
If requeue option sets in true then acknowledgement method sets to REJECT.
### Install (Lua>= 5.1)
1. First step
```sh
luarocks install amqp-client
```
2. Second Step
Replace /usr/local/share/lua/5.1/amqp/init.lua
```sh
git clone https://github.com/Zoviet/amqp-client
cd amqp-client
sudo cp init.lua /usr/local/share/lua/5.1/amqp/init.lua
```
or just copy ampq to your project.
### General usage
Consumer config example:
```
local args = {}
args['x-dead-letter-exchange'] = 'test.retry'
args['x-message-ttl'] = 15000
local ctx = amqp:new({
role = 'consumer',
exchange = 'test',
routing_key = 'test',
queue = 'sms',
ssl = false,
user = 'guest',
password = 'guest',
virtual_host = '/',
no_ack = false,
durable = true,
auto_delete = false,
consumer_tag = '',
requeue = false,
multiple = false,
exclusive = false,
properties = {},
arguments = args
})
```
Consumer loop example:
```
local ok, err
ok , err = ctx:connect(host, port)
if not ok then
error('could not connect'..err)
end
ok, err = ctx:setup() -- because of this we need to use consume_loop()
if not ok then
error('could not setup: '..err)
end
ok, err = ctx:prepare_to_consume() -- this has to be right after setup()
if not ok then
error('could not prepare to consume: '..err)
end
local callback = function(f)
print(f.frame.routing_key)
end
ok, err = ctx:consume_loop(callback)
print(ok, err)
```
Message rejected when callback breaks with error:
```
local callback = function(f)
print(f.frame.routing_key)
error('test error')
end
```
Logging:
```
local logger = require "amqp.logger"
logger.set_level(4) -- logging level for print : ERR = 4 INFO = 7 DEBUG = 8
```
Publisher loop example:
```
local ctx
local function connect()
ctx = amqp:new({
role = 'producer',
exchange = 'missed',
routing_key = 'missed',
ssl = false,
user = 'guest',
password = 'guest',
virtual_host = '/',
no_ack = false,
channel = 2,
durable = true,
auto_delete = true,
consumer_tag = '',
exclusive = false,
properties = {}
})
local ok1, err = ctx:connect(host, port)
if err then
log.error(err)
ctx:close()
wait(5)
connect()
end
local ok, setup_err = ctx:setup()
if setup_err then
log.error(setup_err)
ctx:teardown()
wait(15)
connect()
end
log.info('Connected')
return ok
end
local function publish()
while (true) do
-- Do something
local publish_ok, publish_err = ctx:publish(message)
if not publish_ok then
log.error('Publish error : '..publish_err)
ctx:teardown()
ctx:close()
wait(1)
connect()
end
end
end
if connect() then
publish()
else
ctx:close()
error('Not connect to RabbitMq')
end
ctx:teardown()
```
This library is already tested with RabbitMQ and should work with any other AMQP 0.9.1 broker, follow this [Wiki Documentation](https://github.com/gsdenys/amqp-client/wiki) to know how to use this library.
## Develop
To facilitate to create the lua development environment, was created a [Lua Vagrant Image](https://github.com/gsdenys/lua-vagrant). Use it to gain time.
Case you already have your environment done, just clone this repository and start working. Other else, execute the following steps ([Vagrant](https://www.vagrantup.com) need to be installed)
### Prepare the Environmen
First of all you need to clone the [lua-vagrant](https://github.com/gsdenys/lua-vagrant) environment executing this command:
```sh
git clone https://github.com/gsdenys/vagrant-lua.git
cd vagrant-lua
```
As we'll install the [RabbitMQ](https://www.rabbitmq.com), is very interesting expose the ports to be accessed through the host, this way we can easilly access the administrator enviromnent in your browser. So, to to this, you need to locate the code below at the __Vagrantfile__
```sh
###
# add the network fowarded port here
# ex: config.vm.network "forwarded_port", guest: 15672, host: 8080
###
# config.vm.network "forwarded_port", guest: 8080, host: 8080
```
and insert the the next 2 lines after that:
```sh
config.vm.network "forwarded_port", guest: 15672, host: 8080
config.vm.network "forwarded_port", guest: 672, host: 672
```
So, let's clone this repository inside [lua-vagrant](https://github.com/gsdenys/lua-vagrant) project and start it.
```sh
git clone https://github.com/gsdenys/amqp-client.git
vagrant up #it takes a lot of time
vagrant ssh
```
Now, you already are inside the [lua-vagrant](https://github.com/gsdenys/lua-vagrant) VM. Then you need to install the [RabbitMQ](https://www.rabbitmq.com).
```sh
#install RabbitMQ
wget https://bit.ly/2Ycybe8 -O install.sh
sh install.sh
```
Your environment is ready now. the amqp-client project is in the _/lua/amqp-client_ directory. Go to the project and start to use it.
```sh
#go to amqp-client source code
cd /lua/amqp-client
```
### Building
This library have some requirements shown below. If you're using the [Lua Vagrant Image](https://github.com/gsdenys/vagrant-lua) just take care about the busted lib, the others is already done.
1. LuaJIT >= 2.1
2. busted 2.0 (Testing framework)
3. luabitop (if you are using lua 5.1)
To install busted lib >= 2.1 execute the following command:
```sh
luarocks install busted
```
you need to use the lua socket. Install it with this command:
```sh
luarocks install luasocket
```
After requirements solved, you can run the unit tests with busted using the following command:
```sh
busted
```
Wow you are ready to build the library executing following command::
```sh
luarocks make
```
the output should be like this:
amqp-client 1.0.0-1 is now installed in /usr/local (license: Apache 2.0)
### Examples
The examples needs some dependencies that can be solved through the following commands:
```sh
luarocks install inspect
luarocks install lua-resty-uuid
luarocks install argparse
```
Beyond dependences, the examples depends on build of this library. Other way to execute the examples less building the library is import this library from luarocks. You can do this executing this command.
```sh
luarocks install amqp-client
```

1
_config.yml Normal file
View File

@ -0,0 +1 @@
theme: jekyll-theme-cayman

37
amqp-client-1.3.0-1.rockspec Normal file
View File

@ -0,0 +1,37 @@
package = "amqp-client"
version = "1.3.0-1"
source = {
url = "git://github.com/gsdenys/amqp-client.git",
branch = "v1.3.0-1"
}
description = {
summary = "Lua AMQP 0.9.1 client",
detailed = [[
A pure Lua Client for AMQP 0.9.1.
This library is already tested with RabbitMQ and should work with any other AMQP 0.9.1 broker and can be used with LuaJIT and does not have to be used only in OpenResty.
Developed by: Meng Zhang, Aleksey Konovkin, 4mig4 and Denys Santos.
]],
homepage = "https://github.com/gsdenys/amqp-client",
maintainer = "Denys G. Santos <gsdenys@gmail.com>",
license = "Apache 2.0"
}
dependencies = {
"lua >= 5.1"
}
build = {
type = "builtin",
modules = {
-- ['amqp'] = "src/amqp/init.lua",
['amqp.buffer'] = "src/amqp/buffer.lua",
['amqp.consts'] = "src/amqp/consts.lua",
['amqp.frame'] = "src/amqp/frame.lua",
-- ['amqp.logger'] = "src/amqp/logger.lua",
['amqp'] = 'src/amqp/amqp.lua'
['amqp.context'] = "src/amqp/context.lua",
['amqp.socket.lua'] = "src/amqp/socket/lua.lua",
['amqp.log.simple'] = "src/amqp/log/simple.lua"
}
}

386
amqp/buffer.lua Normal file
View File

@ -0,0 +1,386 @@
--
-- Copyright (C) 2016 Meng Zhang @ Yottaa,Inc
-- Copyright (C) 2018 4mig4
--
local logger = require('amqp.logger')
local bit = require('bit')
local band = bit.band
local bor = bit.bor
local lshift = bit.lshift
local rshift = bit.rshift
local tohex = bit.tohex
local concat = table.concat
local sub = string.sub
local byte = string.byte
local char = string.char
local format = string.format
local ok, new_tab = pcall(require, "table.new")
if not ok or type(new_tab) ~= "function" then
new_tab = function (_,_) return {} end
end
local buffer = {}
local mt = { __index = buffer }
function buffer.new(b,pos)
return setmetatable( { buffer_ = b or "", pos_ = pos or 1 }, mt)
end
function buffer:hex_dump()
local len = #self.buffer_
local bytes = new_tab(len, 0)
for i = 1, len do
bytes[i] = tohex(byte(self.buffer_, i), 2)
end
return concat(bytes, " ")
end
function buffer:get_i8()
local b = byte(self.buffer_, self.pos_)
self.pos_ = self.pos_ + 1
return b
end
function buffer:get_bool()
local v = self:get_i8()
return v and v ~= 0
end
function buffer:get_i16()
local a0, a1 = byte(self.buffer_, self.pos_, self.pos_ + 1)
local r = bor(a1, lshift(a0, 8))
self.pos_ = self.pos_ + 2
return r
end
function buffer:get_i24()
local a0, a1, a2 = byte(self.buffer_, self.pos_, self.pos_ + 2)
self.pos_ = self.pos_ + 3
return bor(a2,
lshift(a1, 8),
lshift(a0, 16))
end
function buffer:get_i32()
local a0, a1, a2, a3 = byte(self.buffer_, self.pos_, self.pos_ + 3)
self.pos_ = self.pos_ + 4
return bor(a3,
lshift(a2, 8),
lshift(a1, 16),
lshift(a0, 24))
end
function buffer:get_i64()
local a, b, c, d, e, f, g, h = byte(self.buffer_, self.pos_, self.pos_ + 7)
self.pos_ = self.pos_ + 8
local lo = bor(h, lshift(g, 8), lshift(f, 16), lshift(e, 24))
local hi = bor(d, lshift(c, 8), lshift(b, 16), lshift(a, 24))
return lo + hi * 4294967296
end
function buffer:get_short_string()
local length = self:get_i8()
local tail = self.pos_+length-1
local s = sub(self.buffer_,self.pos_,tail)
self.pos_ = tail + 1
return s
end
function buffer:get_long_string()
local length = self:get_i32()
local tail = self.pos_+length-1
local s = sub(self.buffer_,self.pos_,tail)
self.pos_ = tail + 1
return s
end
function buffer:get_decimal()
local scale = self:get_i8()
local value = self:get_i32()
local d = {scale = scale, value = value}
return d
end
function buffer:get_f32()
return self:get_i32()
end
function buffer:get_f64()
return self:get_i64()
end
function buffer:get_timestamp()
return self:get_i64()
end
function buffer:get_field_array()
local size = self:get_i32()
logger.dbg("[array] size: " .. size)
local a = {}
local p = self.pos_
while size > 0 do
local f = self:field_value()
a[#a+1] = f
size = size - (self.pos_ - p)
p = self.pos_
end
return a
end
function buffer:get_field_table()
local size = self:get_i32()
local r = {}
local p = self.pos_
local k,v
while size > 0 do
k = self:get_short_string()
v = self:field_value()
size = size - self.pos_ + p
p = self.pos_
r[k] = v
end
return r
end
function buffer:put_i8(i)
self.buffer_ = self.buffer_ .. char(band(i,0x0ff))
end
function buffer:put_bool(b)
local v = 0
if b and b ~= 0 then
v = 1
end
self:put_i8(v)
end
function buffer:put_i16(i)
self.buffer_ = self.buffer_ ..
char(rshift(band(i,0xff00),8)) ..
char(band(i,0x0ff))
end
function buffer:put_i32(i)
self.buffer_ = self.buffer_ ..
char(rshift(band(i,0xff000000),24)) ..
char(rshift(band(i, 0x00ff0000),16)) ..
char(rshift(band(i, 0x0000ff00),8)) ..
char(band(i, 0x000000ff))
end
function buffer:put_i64(i)
-- rshift has not support for 64bit?
-- side effect is that it will rotate for shifts bigger than 32bit
local hi = band(i/4294967296,0x0ffffffff)
local lo = band(i, 0x0ffffffff)
self:put_i32(hi)
self:put_i32(lo)
end
function buffer:put_f32(i)
self:put_i32(i)
end
function buffer:put_f64(i)
self:put_i64(i)
end
function buffer:put_timestamp(i)
self:put_i64(i)
end
function buffer:put_decimal(d)
self:put_i8(d.scale)
self:put_i32(d.value)
end
function buffer:put_short_string(s)
local len = #s
self:put_i8(len)
self.buffer_ = self.buffer_ .. s
end
function buffer:put_long_string(s)
local len = #s
self:put_i32(len)
self.buffer_ = self.buffer_ .. s
end
function buffer:put_payload(payload)
self.buffer_ = self.buffer_ .. payload
end
local function is_array(t)
local i = 0
for _ in pairs(t) do
i = i + 1
if t[i] == nil then return false end
end
return true
end
function buffer:put_field_array(a)
local b = buffer.new()
for i = 1, #a do
b:put_field_value(a[i])
end
self:put_i32(#b.buffer_)
self:put_payload(b.buffer_)
end
function buffer:put_field_table(tab)
local b = buffer.new()
for k,v in pairs(tab) do
b:put_short_string(k)
b:put_field_value(v)
end
self:put_i32(#b.buffer_)
self:put_payload(b.buffer_)
end
local fields_ = {
t = {
r = function (self)
local b = self:get_i8()
return b ~= 0
end,
w = function (self,val)
local b = 0
if val ~= 0 then
b = 1
end
self:put_i8(b)
end
},
b = {
r = buffer.get_i8,
w = buffer.put_i8
},
B = {
r = buffer.get_i8,
w = buffer.put_i8
},
U = {
r = buffer.get_i16,
w = buffer.put_i16
},
u = {
r = buffer.get_i16,
w = buffer.put_i16
},
I = {
r = buffer.get_i32,
w = buffer.put_i32
},
i = {
r = buffer.get_i32,
w = buffer.put_i32
},
L = {
r = buffer.get_i64,
w = buffer.put_i64
},
l = {
r = buffer.get_i64,
w = buffer.put_i64
},
f = {
r = buffer.get_f32,
w = buffer.put_f32
},
d = {
r = buffer.get_f64,
w = buffer.put_f64
},
D = {
r = buffer.get_decimal,
w = buffer.put_decimal
},
s = {
r = buffer.get_short_string,
w = buffer.put_short_string
},
S = {
r = buffer.get_long_string,
w = buffer.put_long_string
},
A = {
r = buffer.get_field_array,
w = buffer.put_field_array
},
T = {
r = buffer.get_timestamp,
w = buffer.put_timestamp
},
F = {
r = buffer.get_field_table,
w = buffer.put_field_table
},
x = {
r = buffer.get_long_string,
w = buffer.put_long_string
},
V = {
r = function()
return nil
end,
w = nil
}
}
function buffer:field_value()
local typ = self:get_i8()
local codec = fields_[char(typ)]
if not codec then
local err = format("codec[%d] not found.",typ)
logger.error("[field_value] " .. err)
return nil,err
end
return codec.r(self)
end
function buffer:put_field_value(value)
-- FIXME: to detect the type of the value
local t = type(value)
local typ = nil
if t == 'number' then
typ = 'I' -- assume to be i32
elseif t == 'boolean' then
typ = 't'
elseif t == 'string' then
typ = 'S'
elseif t == 'table' then
typ = 'F'
if is_array(value) then
typ = 'A'
end
end
-- wire the type
self:put_i8(byte(typ))
local codec = fields_[typ]
if not codec then
local err = format("codec[%d] not found.",typ)
logger.error("[field_value] " .. err)
return nil,err
end
codec.w(self,value)
end
function buffer:payload()
return self.buffer_
end
return buffer

174
amqp/consts.lua Normal file
View File

@ -0,0 +1,174 @@
--
-- Copyright (C) 2016 Meng Zhang @ Yottaa,Inc
-- Copyright (C) 2018 4mig4
--
--
-- Constants definition
--
local amqp = {
DEFAULT_CHANNEL = 0,
DEFAULT_FRAME_SIZE = 131072,
DEFAULT_MAX_CHANNELS = 65535,
DEFAULT_HEARTBEAT = 30,
PROTOCOL_VERSION_MAJOR = 0,
PROTOCOL_VERSION_MINOR = 9,
PROTOCOL_VERSION_REVISION = 1,
PROTOCOL_PORT = 5672,
PROTOCOL_SSL_PORT = 5671,
state = {
CLOSED = 0,
ESTABLISHED = 1,
CLOSE_WAIT = 2,
},
frame = {
METHOD_FRAME = 1,
HEADER_FRAME = 2,
BODY_FRAME = 3,
HEARTBEAT_FRAME = 8,
FRAME_MIN_SIZE = 4096,
FRAME_END = 0xCE
},
method = {
connection = {
START = 0x0A,
START_OK = 0x0B,
SECURE = 0x14,
SECURE_OK = 0x15,
TUNE = 0x1E,
TUNE_OK = 0x1F,
OPEN = 0x28,
OPEN_OK = 0x29,
CLOSE = 0x32,
CLOSE_OK = 0x33,
BLOCKED = 0x3C,
UNBLOCKED = 0x3D,
},
channel = {
OPEN = 0x0A,
OPEN_OK = 0x0B,
FLOW = 0x14,
FLOW_OK = 0x15,
CLOSE = 0x28,
CLOSE_OK = 0x29,
},
exchange = {
DECLARE = 0x0A,
DECLARE_OK = 0x0B,
DELETE = 0x14,
DELETE_OK = 0x15,
BIND = 0x1E,
BIND_OK = 0x1F,
UNBIND = 0x28,
UNBIND_OK = 0x33,
},
queue = {
DECLARE = 0x0A,
DECLARE_OK = 0x0B,
BIND = 0x14,
BIND_OK = 0x15,
PURGE = 0x1E,
PURGE_OK = 0x1F,
DELETE = 0x28,
DELETE_OK = 0x29,
UNBIND = 0x32,
UNBIND_OK = 0x33,
},
basic = {
QOS = 0x0A,
QOS_OK = 0x0B,
CONSUME = 0x14,
CONSUME_OK = 0x15,
CANCEL = 0x1E,
CANCEL_OK = 0x1F,
PUBLISH = 0x28,
RETURN = 0x32,
DELIVER = 0x3C,
GET = 0x46,
GET_OK = 0x47,
GET_EMPTY = 0x48,
ACK = 0x50,
REJECT = 0x5A,
RECOVER_ASYNC = 0x64,
RECOVER = 0x6E,
RECOVER_OK = 0x6F,
NACK = 0x78,
},
tx = {
SELECT = 0x0A,
SELECT_OK = 0x0B,
COMMIT = 0x14,
COMMIT_OK = 0x15,
ROLLBACK = 0x1E,
ROLLBACK_OK = 0x1F,
},
confirm = {
SELECT = 0x0A,
SELECT_OK = 0x0B
}
},
class = {
CONNECTION = 0x000A,
CHANNEL = 0x0014,
EXCHANGE = 0x0028,
QUEUE = 0x0032,
BASIC = 0x003C,
TX = 0x005A,
CONFIRM = 0x0055
},
flag = {
CONTENT_TYPE = 0x8000,
CONTENT_ENCODING = 0x4000,
HEADERS = 0x2000,
DELIVERY_MODE = 0x1000,
PRIORITY = 0x0800,
CORRELATION_ID = 0x0400,
REPLY_TO = 0x0200,
EXPIRATION = 0x0100,
MESSAGE_ID = 0x0080,
TIMESTAMP = 0x0040,
TYPE = 0x0020,
USER_ID = 0x0010,
APP_ID = 0x0008,
RESERVED1 = 0x0004
},
err = {
REPLY_SUCCESS = 200,
CONTENT_TOO_LARGE = 311,
NO_ROUTE = 312,
NO_CONSUMERS = 313,
CONNECTION_FORCED = 320,
INVALID_PATH = 402,
ACCESS_REFUSED = 403,
NOT_FOUND = 404,
RESOURCE_LOCKED = 405,
PRECONDITION_FAILED = 406,
FRAME_ERROR = 501,
SYNTAX_ERROR = 502,
COMMAND_INVALID = 503,
CHANNEL_ERROR = 504,
UNEXPECTED_FRAME = 505,
RESOURCE_ERROR = 506,
NOT_ALLOWED = 530,
NOT_IMPLEMENTED = 540,
INTERNAL_ERROR = 541
},
PRODUCT = "amqp-client",
VERSION = "1.0.3",
COPYRIGHT = "Copyright (c) 2016,2017,2018,2019 Meng Zhang @Yottaa, @4mig4, @gsdenys",
LOCALE = "en_US",
MECHANISM_PLAIN = "PLAIN"
}
return amqp

1592
amqp/frame.lua Normal file
View File

File diff suppressed because it is too large Load Diff

1060
amqp/init.lua Executable file
View File

File diff suppressed because it is too large Load Diff

95
amqp/logger.lua Normal file
View File

@ -0,0 +1,95 @@
--
-- Copyright (C) 2016 Meng Zhang @ Yottaa,Inc
-- Copyright (C) 2018 4mig4
--
-- logger
--
local logger = {}
-- logging scaffold
local log
if _G.ngx and _G.ngx.log then
log = _G.ngx.log
else
log = print
end
-- logging level for print
local ERR = 4
local INFO = 7
local DEBUG = 8
-- ngx.log requires a number to indicate the logging level
if _G.ngx then
ERR = _G.ngx.ERR
INFO = _G.ngx.INFO
DEBUG = _G.ngx.DEBUG
end
local level_ = INFO
local function to_string(v_)
if v_ == nil then
return ""
end
if type(v_) ~= "table" then
return tostring(v_)
end
local s = "["
for k,v in pairs(v_) do
if k ~= nil then
s = s .. to_string(k) .. ":"
end
if v ~= nil then
s = s .. to_string(v)
end
s = s .. " "
end
s = s .. "]"
return s
end
local function va_table_to_string(tbl)
local res = ""
for _,v in pairs(tbl) do
res = res .. to_string(v) .. "\t"
end
return res
end
function logger.set_level(level)
level_ = level
end
function logger.error(...)
if level_ < ERR then
return
end
log(ERR, va_table_to_string({...}))
end
function logger.info(...)
if level_ < INFO then
return
end
log(INFO, va_table_to_string({...}))
end
function logger.dbg(...)
if level_ ~= DEBUG then
return
end
log(DEBUG, va_table_to_string({...}))
end
function logger.is_debug_enabled()
return level_ == DEBUG
end
return logger

21
examples/connection/connect.lua Normal file
View File

@ -0,0 +1,21 @@
-- Import the AMQP Client Library
local amqp = require("amqp")
-- Define the connection URL and Port
local host = "127.0.0.1"
local port = 5672
-- Pepare to connect to AMQP server
local ctx = amqp:new({
role = 'producer',
exchange = 'amq.topic',
routing_key = 'xpto',
ssl = false,
user = 'admin',
password = 'admin'
})
local ok, err = ctx:connect(host, port)
print("Connection Created:", ok, "Errors:", err)
local ok, err = ctx:setup()
print("Connection setup:", ok, "Errors:", err)

27
examples/exchange/bind.lua Normal file
View File

@ -0,0 +1,27 @@
-- Import the AMQP Client Library
local amqp = require("amqp")
-- Define the connection URL and Port
local host = "127.0.0.1"
local port = 5672
-- Connect to AMQP server
local ctx = amqp:new({
role = "producer",
user = "admin",
password = "admin"
})
local ok, err = ctx:connect("127.0.0.1",5672)
local ok, err = ctx:setup()
-- Bind an the amq.direct to the 'drx.xpto'exchange with
-- the routing key 'potato' (the 'drx.xpto' exchange needs to exist)
local ok, err = amqp.exchange_bind(ctx,{
source = "amq.direct",
destination = "drt.xpto",
routing_key = "potato"
})
-- Finalize proccess and close connection
ctx:teardown()
ctx:close()

28
examples/exchange/create.lua Normal file
View File

@ -0,0 +1,28 @@
-- Import the AMQP Client Library
local amqp = require("amqp")
-- Define the connection URL and Port
local host = "127.0.0.1"
local port = 5672
-- Connect to AMQP server
local ctx = amqp:new({
role = "producer",
user = "admin",
password = "admin"
})
local ok, err = ctx:connect("127.0.0.1",5672)
local ok, err = ctx:setup()
-- Create a new direct exchange with name 'topic.xpto',
-- durable and with bit auto delete as true
local ok, err = amqp.exchange_declare(ctx,{
exchange = "drt.xpto",
durable = true,
auto_delete = true,
typ = "direct"
})
-- Finalize proccess and close connection
ctx:teardown()
ctx:close()

24
examples/exchange/delete.lua Normal file
View File

@ -0,0 +1,24 @@
-- Import the AMQP Client Library
local amqp = require("amqp")
-- Define the connection URL and Port
local host = "127.0.0.1"
local port = 5672
-- Connect to AMQP server
local ctx = amqp:new({
role = "producer",
user = "admin",
password = "admin"
})
local ok, err = ctx:connect("127.0.0.1",5672)
local ok, err = ctx:setup()
-- Delete the 'drx.xpto'exchange
local ok, err = amqp.exchange_delete(ctx,{
exchange = "drt.xpto",
})
-- Finalize proccess and close connection
ctx:teardown()
ctx:close()

27
examples/exchange/unbind.lua Normal file
View File

@ -0,0 +1,27 @@
-- Import the AMQP Client Library
local amqp = require("amqp")
-- Define the connection URL and Port
local host = "127.0.0.1"
local port = 5672
-- Connect to AMQP server
local ctx = amqp:new({
role = "producer",
user = "admin",
password = "admin"
})
local ok, err = ctx:connect("127.0.0.1",5672)
local ok, err = ctx:setup()
-- Unbind an the amq.direct to the 'drx.xpto'exchange with
-- the routing key 'potato'
local ok, err = amqp.exchange_unbind(ctx,{
source = "amq.direct",
destination = "drt.xpto",
routing_key = "potato"
})
-- Finalize proccess and close connection
ctx:teardown()
ctx:close()

32
examples/message/consumer.lua Normal file
View File

@ -0,0 +1,32 @@
-- Import the AMQP Client Library
local amqp = require "amqp"
local inspect = require('inspect')
-- Define the connection URL and Port
local host = "127.0.0.1"
local port = 5672
-- Connect to AMQP server
local ctx = amqp:new({
role = 'consumer',
exchange = '',
queue = 'test',
ssl = false,
user = 'admin',
password = 'admin',
no_ack = false,
durable = true,
auto_delete = false,
consumer_tag = '',
exclusive = false,
properties = {}
})
local ok, err = ctx:connect(host, port)
-- Define a callback for consumer
ctx.opts.callback = function(f)
print(inspect(f))
end
-- Start consume loop
ok, err = ctx:consume()

43
examples/message/consumer_loop.lua Normal file
View File

@ -0,0 +1,43 @@
local amqp = require "amqp"
local inspect = require('inspect')
local host = "127.0.0.1"
local port = 5672
local ctx = amqp:new({
role = 'consumer',
exchange = '',
queue = 'test',
ssl = false,
user = 'admin',
password = 'admin',
no_ack = false,
durable = true,
auto_delete = false,
consumer_tag = '',
exclusive = false,
properties = {}
})
local ok, err
ok , err = ctx:connect(host, port)
if not ok then
error('could not connect'..err)
end
ok, err = ctx:setup() -- because of this we need to use consume_loop()
if not ok then
error('could not setup: '..err)
end
ok, err = ctx:prepare_to_consume() -- this has to be right after setup()
if not ok then
error('could not prepare to consume: '..err)
end
local callback = function(f)
print('---f---',inspect(f))
end
ok, err = ctx:consume_loop(callback)
print(ok, err)

30
examples/message/publisher.lua Normal file
View File

@ -0,0 +1,30 @@
local amqp = require "amqp"
local logger = require "amqp.logger"
local uuid = require('resty.uuid')
local host = "127.0.0.1"
local port = 5672
logger.set_level(7)
local ctx = amqp:new({
role = 'producer',
exchange = 'amq.topic',
routing_key = 'test',
ssl = false,
user = 'admin',
password = 'admin',
no_ack = false,
durable = true,
auto_delete = true,
consumer_tag = '',
exclusive = false,
properties = {}
})
local ok1, err1 = ctx:connect(host, port)
local ok2, err2 = ctx:setup()
local ok3, err3 = ctx:publish("Hello world!",{},{correlation_id = uuid.generate()})
ctx:teardown()
ctx:close()

27
examples/queue/bind.lua Normal file
View File

@ -0,0 +1,27 @@
-- Import the AMQP Client Library
local amqp = require("amqp")
-- Define the connection URL and Port
local host = "127.0.0.1"
local port = 5672
-- Connect to AMQP server
local ctx = amqp:new({
role = "producer",
user = "admin",
password = "admin"
})
local ok, err = ctx:connect("127.0.0.1",5672)
local ok, err = ctx:setup()
-- Bind an the amq.direct to the 'xpto'queue with
-- the routing key 'potato'
local ok, err = amqp.queue_bind(ctx,{
exchange = "amq.direct",
queue = "xpto",
routing_key = "potato"
})
-- Finalize proccess and close connection
ctx:teardown()
ctx:close()

28
examples/queue/create.lua Normal file
View File

@ -0,0 +1,28 @@
-- Import the AMQP Client Library
local amqp = require("amqp")
-- Define the connection URL and Port
local host = "127.0.0.1"
local port = 5672
-- Connect to AMQP server
local ctx = amqp:new({
role = "producer",
user = "admin",
password = "admin"
})
local ok, err = ctx:connect("127.0.0.1",5672)
local ok, err = ctx:setup()
-- Create a new queue with name 'xpto', durable and
-- with bit auto delete as true
local ok, err = amqp.queue_declare(ctx,{
queue = "xpto",
durable = true,
auto_delete = false,
typ = "direct"
})
-- Finalize proccess and close connection
ctx:teardown()
ctx:close()

24
examples/queue/delete.lua Normal file
View File

@ -0,0 +1,24 @@
-- Import the AMQP Client Library
local amqp = require("amqp")
-- Define the connection URL and Port
local host = "127.0.0.1"
local port = 5672
-- Connect to AMQP server
local ctx = amqp:new({
role = "producer",
user = "admin",
password = "admin"
})
local ok, err = ctx:connect("127.0.0.1",5672)
local ok, err = ctx:setup()
-- Delete the 'xpto' queue
local ok, err = amqp.queue_delete(ctx,{
queue = "xpto",
})
-- Finalize proccess and close connection
ctx:teardown()
ctx:close()

27
examples/queue/unbind.lua Normal file
View File

@ -0,0 +1,27 @@
-- Import the AMQP Client Library
local amqp = require("amqp")
-- Define the connection URL and Port
local host = "127.0.0.1"
local port = 5672
-- Connect to AMQP server
local ctx = amqp:new({
role = "producer",
user = "admin",
password = "admin"
})
local ok, err = ctx:connect("127.0.0.1",5672)
local ok, err = ctx:setup()
-- Unbind the amq.direct to the 'xpto' queue with
-- the routing key 'potato'
local ok, err = amqp.queue_unbind(ctx,{
exchange = "amq.direct",
queue = "xpto",
routing_key = "potato"
})
-- Finalize proccess and close connection
ctx:teardown()
ctx:close()

180
examples/rpc/amqp_rpc_client.lua Executable file
View File

@ -0,0 +1,180 @@
#!/usr/bin/luajit
-- @author 4mig4 <4mig4-github@gmail.com>
-- @release 1.0.0
-- @licence ISC
--
---------------------------------------------------------------------------------
-- Copyright (c) 2018 4mig4 <4mig4-github@gmail.com>
--
-- Permission to use, copy, modify, and distribute this software for any
-- purpose with or without fee is hereby granted, provided that the above
-- copyright notice and this permission notice appear in all copies.
--
-- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
-- WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
-- MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
-- ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
-- WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
-- ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
-- OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
---------------------------------------------------------------------------------
-- PING PONG (Client) example
-- https://www.rabbitmq.com/direct-reply-to.html
---------------------------------------------------------------------------------
-- Requires
local amqp = require ('amqp')
local uuid = require('resty.uuid')
local inspect = require('inspect')
local argparse = require('argparse')
--local cqueues = require('cqueues')
---------------------------------------------------------------------------------
-- Main
local parser = argparse()
--
-- Instantiate a context
--
local ctx = amqp:new({role = "consumer",
exchange = '',
queue = 'amq.rabbitmq.reply-to',
routing_key = 'test',
ssl = false,
user = "guest",
password = "guest",
no_ack = true,
durable = true,
auto_delete = true,
consumer_tag = '',
exclusive = false,
properties = {}
})
--
-- Define callback function that will process the answer (see below)
--
ctx.opts.callback =
function(f)
print('---f---',inspect(f))
if f.body == 'pong' then
print('received: pong')
local payload = 'ping'
local correlation_id = uuid.generate()
local properties = {
reply_to = 'amq.rabbitmq.reply-to',
content_type = 'application/json',
content_encoding = 'utf-8',
correlation_id = correlation_id,
delivery_mode = 2,
headers = { ['api-version'] = 1, -- custome headers
correlation_id = correlation_id }
}
local ok, err = ctx:publish(payload, ctx.opts, properties)
if ok then
print('replied: ping')
else
error('could not publish: '..err)
end
end
end
local function rpc_client(args)
local host = '127.0.0.1'
local port = 5672
if ctx.opts.user and args.user then ctx.opts.user = args.user end
if ctx.opts.password and args.password then ctx.opts.password = args.password end
if args.ssl then ctx.opts.ssl = args.ssl end
if args.host then host = args.host end
if args.port then port = args.port end
local ok, err
ok , err = ctx:connect(host,port)
if not ok then
error('could not connect'..err)
end
--
-- Setup connection to AMQP server
--
ok, err = ctx:setup() -- because of this we need to use consume_loop()
if not ok then
error('could not setup: '..err)
end
--
-- Prepare to consume
--
ok, err = ctx:prepare_to_consume() -- this has to be right after setup()
if not ok then
error('could not prepare to consume: '..err)
end
--
-- Publish ( AKA send command to PING-PONG server in our case )
--
local correlation_id = uuid.generate()
local properties = {
reply_to = 'amq.rabbitmq.reply-to',
content_type = 'application/json',
content_encoding = 'utf-8',
correlation_id = correlation_id,
delivery_mode = 2,
headers = { ['api-version'] = 1,
correlation_id = correlation_id }
}
local payload = 'ping'
ok, err = ctx:publish(payload, ctx.opts, properties)
if not ok then
error('could not publish: '..err)
end
print('sent: ping')
--
-- Wait for answer
--
-- Enter consume loop
-- callback will be called only on reception of BODY_FRAME.
-- setup() and prepare_to_consume() calls are needed because were are using
-- consume_loop(callback) method and not consume() method.
ok = ctx:consume_loop(ctx.opts.callback)
if not ok then
error('could not consume loop: ')
end
end
-- Argument handling ---------------------------------------------------------
parser:option('-H --host','rabbitmq hostname', '127.0.0.1')
parser:option('-P --port','rabbitmq port', 5672)
parser:option('-u --user','rabbitmq username', 'guest')
parser:option('-p --password','rabbitmq password', 'guest')
parser:flag('-s --ssl','enable ssl')
local args = parser:parse()
rpc_client(args)

139
examples/rpc/amqp_rpc_server.lua Executable file
View File

@ -0,0 +1,139 @@
#!/usr/bin/luajit
-- @author 4mig4 <4mig4-github@gmail.com>
-- @release 1.0.0
-- @licence ISC
--
---------------------------------------------------------------------------------
-- Copyright (c) 2018 4mig4 <4mig4-github@gmail.com>
--
-- Permission to use, copy, modify, and distribute this software for any
-- purpose with or without fee is hereby granted, provided that the above
-- copyright notice and this permission notice appear in all copies.
--
-- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
-- WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
-- MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
-- ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
-- WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
-- ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
-- OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
---------------------------------------------------------------------------------
-- PING-PONG (Server) example
-- https://www.rabbitmq.com/direct-reply-to.html
---------------------------------------------------------------------------------
-- Requires
local amqp = require('amqp')
local inspect = require('inspect')
local argparse = require('argparse')
--local cqueues = require('cqueues')
---------------------------------------------------------------------------------
-- Main
local parser = argparse()
--
-- Instantiates context
--
local ctx = amqp:new({role = 'consumer',
exchange = '',
queue = 'test',
routing_key = '',
ssl = false,
user = 'guest',
password = 'guest',
no_ack = false,
durable = true,
auto_delete = true,
consumer_tag = '',
exclusive = false,
properties = {}
}
)
--
-- Definition of the callback function
--
ctx.opts.callback = function(f)
print('-- f --', inspect(f))
if f.body == 'ping' then
print('received: ping')
local response = 'pong'
local correlation_id
if f.properties.reply_to then
ctx.opts.routing_key = f.properties.reply_to
end
if f.properties.correlation_id then
correlation_id = f.properties.correlation_id
end
local properties = { correlation_id = correlation_id,
delivery_mode = 2,
headers = { ['api-version'] = 1, -- custom headers
correlation_id = correlation_id }
}
local ok , err = ctx:publish(response, ctx.opts, properties)
if ok then
print('sent: pong')
return true
else
return nil, 'could not publish' .. inspect(err)
end
end
end
local function rpc_main(args)
local host = '127.0.0.1'
local port = 5672
--
-- Connect to AMQP server (broker)
--
if ctx.opts.user and args.user then ctx.opts.user = args.user end
if ctx.opts.password and args.password then ctx.opts.password = args.password end
if args.ssl then ctx.opts.ssl = args.ssl end
if args.host then host = args.host end
if args.port then port = args.port end
local ok, err
ok = ctx:connect(host, port)
if not ok then
error('failed to connect')
end
--
-- Enter consume loop
--
-- Callback will be called only on reception of BODY_FRAME.
-- No need for setup() or prepare_to_consume() calls because were are using
-- consume() and not consume_loop(callback), also note that consume() the
-- callback function is passed thru the ctx context and not as an argument
-- of the method like in in consume_loop()
ok, err = ctx:consume()
if not ok then
error('consume failed: '..err)
end
end
-- Argument handling ---------------------------------------------------------
parser:option('-H --host','rabbitmq hostname', '127.0.0.1')
parser:option('-P --port','rabbitmq port', 5672)
parser:option('-u --user','rabbitmq username', 'guest')
parser:option('-p --password','rabbitmq password', 'guest')
parser:flag('-s --ssl','enable ssl')
local args = parser:parse()
rpc_main(args)

58
examples/rpc/rpc_client.lua Normal file
View File

@ -0,0 +1,58 @@
local amqp = require "amqp"
local uuid = require('resty.uuid')
local inspect = require('inspect')
local host = "127.0.0.1"
local port = 5672
local producer = amqp:new({
role = 'producer',
exchange = 'amq.topic',
routing_key = 'test',
ssl = false,
user = 'admin',
password = 'admin',
no_ack = false,
durable = true,
auto_delete = true,
consumer_tag = '',
exclusive = false,
properties = {}
})
uid = uuid.generate()
local ok1, err1 = producer:connect(host, port)
local ok2, err2 = producer:setup()
local ok3, err3 = producer:publish("Hello world!",{},{correlation_id = uid})
-- ############
local consumer = amqp:new({
role = 'consumer',
exchange = '',
queue = uid..'-queue',
ssl = false,
user = 'admin',
password = 'admin',
no_ack = false,
durable = true,
auto_delete = true,
consumer_tag = '',
exclusive = false,
properties = {}
})
local ok, err
ok , err = consumer:connect(host, port)
if not ok then
error('could not connect'..err)
end
-- register calback
consumer.opts.callback = function(f)
print('-- f --', inspect(f))
end
local ok, err = consumer:consume()

1060
init.lua Executable file
View File

File diff suppressed because it is too large Load Diff

28
spec/basic_spec.lua Normal file
View File

@ -0,0 +1,28 @@
require 'busted.runner'()
describe("[amqp]", function()
describe("client", function()
local logger = require("amqp.logger")
logger.set_level(7)
it("should be able to publish messages", function()
local amqp = require("amqp")
local ctx = amqp:new({
role = 'producer',
exchange = 'amq.topic',
ssl = false,
user = 'admin',
password = 'admin'
})
local ok, err = ctx:connect("127.0.0.1",5672)
assert.truthy(ok)
local ok, err = ctx:setup()
assert.truthy(ok)
local ok, err = ctx:publish('{"Hello": "world!"}')
assert.truthy(ok)
ctx:teardown()
ctx:close()
end)
end)
end)

27
spec/connection_spec.lua Normal file
View File

@ -0,0 +1,27 @@
require 'busted.runner'()
describe("[amqp]", function()
describe("client", function()
local logger = require("amqp.logger")
logger.set_level(7)
it("should be rejected with wrong user name or password", function()
local amqp = require("amqp")
local ctx = amqp:new({
role = 'producer',
exchange = 'amq.topic',
routing_key = 'xpto',
ssl = false,
user = 'admin',
password = 'adminerr'
})
local ok, err = ctx:connect("127.0.0.1",5672)
assert.truthy(ok)
local ok, err = ctx:setup()
-- expects access denied
assert.is_equal(403,err)
assert.falsy(ok)
end)
end)
end)

108
spec/exchange_spec.lua Normal file
View File

@ -0,0 +1,108 @@
require 'busted.runner'()
describe("[amqp]", function()
describe("client", function()
local logger = require("amqp.logger")
logger.set_level(9)
local options = {
role = "producer",
exchange = "amq.topic",
routing_key = 'xpto',
ssl = false,
user = "admin",
password = "admin"
}
it("should be able to declare exchange", function()
local amqp = require("amqp")
local ctx = amqp:new(options)
local ok, err = ctx:connect("127.0.0.1",5672)
assert.truthy(ok)
local ok, err = ctx:setup()
assert.truthy(ok)
local ok, err = amqp.exchange_declare(ctx,{
exchange = "topic.xpto",
passive = false,
durable = true,
internal = false,
auto_delete = true
})
assert.truthy(ok)
ctx:teardown()
ctx:close()
end)
it("should be able to bind exchange", function()
local amqp = require("amqp")
local ctx = amqp:new(options)
local ok, err = ctx:connect("127.0.0.1",5672)
assert.truthy(ok)
local ok, err = ctx:setup()
assert.truthy(ok)
local ok, err = amqp.exchange_bind(ctx,{
source = "amq.topic",
destination = "topic.xpto",
routing_key = "Kiwi"
})
assert.truthy(ok)
ctx:teardown()
ctx:close()
end)
it("should be able to unbind exchange", function()
local amqp = require("amqp")
local ctx = amqp:new(options)
local ok, err = ctx:connect("127.0.0.1",5672)
assert.truthy(ok)
local ok, err = ctx:setup()
assert.truthy(ok)
local ok, err = amqp.exchange_unbind(ctx,{
source = "amq.topic",
destination = "topic.xpto",
routing_key = "Kiwi"
})
assert.truthy(ok)
ctx:teardown()
ctx:close()
end)
it("should be able to delete exchange", function()
local amqp = require("amqp")
local ctx = amqp:new(options)
local ok, err = ctx:connect("127.0.0.1",5672)
assert.truthy(ok)
local ok, err = ctx:setup()
assert.truthy(ok)
local ok, err = amqp.exchange_delete(ctx,{
exchange = "topic.xpto"
})
assert.truthy(ok)
ctx:teardown()
ctx:close()
end)
it("should succeed to delete non existing exchanges", function()
local amqp = require("amqp")
local ctx = amqp:new(options)
local ok, err = ctx:connect("127.0.0.1",5672)
assert.truthy(ok)
local ok, err = ctx:setup()
assert.truthy(ok)
local ok, err = amqp.exchange_delete(ctx,{
exchange = "topic.xpto"
})
assert.truthy(ok)
ctx:teardown()
ctx:close()
end)
end)
end)

17
spec/socket/lua_spec.lua Normal file
View File

@ -0,0 +1,17 @@
require 'busted.runner'()
describe("Lua Socket", function ()
it("Should connect to server", function()
local sock = require "amqp.socket.lua"
sock:connect('httpbin.org', 80)
end)
it("Should send message", function()
local sock = require "amqp.socket.lua"
sock:connect('httpbin.org', 80)
a = sock:send("GET /anything")
assert.not_nil(a)
end)
end)

5
src/amqp.bkp/amqp.lua Normal file
View File

@ -0,0 +1,5 @@
local ctx = require 'amqp.commons.context'
local socket = require 'amqp.socket.selector'

85
src/amqp.bkp/commons/context.lua Normal file
View File

@ -0,0 +1,85 @@
local consts = require 'amqp.consts'
local function new(opts, sock)
local default_ctx = {
connection_state = consts.state.CLOSED,
channel_state = consts.state.CLOSED,
channel_max = consts.DEFAULT_MAX_CHANNELS,
frame_max = consts.DEFAULT_FRAME_SIZE,
mechanism = consts.MECHANISM_PLAIN,
revision = consts.PROTOCOL_VERSION_REVISION,
ssl_ctx = opts.ssl_ctx or nil,
timeout = consts.DEFAULT_TIMEOUT,
major = consts.PROTOCOL_VERSION_MAJOR,
minor = consts.PROTOCOL_VERSION_MINOR,
sock = sock,
opts = opts
}
local ctx = default_ctx
-- Set opts
-- function to set opts inside context. As ssl_ctx is dependent of opts,
-- this function, when called, will overwrite it.
--
-- @param opts connection and socket options
function ctx:set_opts (opts)
self.opts = opts
self.ssl_ctx = opts.ssl_ctx or nil
end
-- Set sock
-- function to set the sock
--
-- @param sock the sock
function ctx:set_sock(sock)
self.sock = sock
end
-- Set timeout
-- Change the system timeou. It's just functional before connect to provider,
-- after connection made, function will have no effect.
--
-- @param timeout the timeout to be set (default 5000)
function ctx:set_timeout(timeout)
self.timeout = timeout or consts.DEFAULT_TIMEOUT
end
-- Set ssl_ctx
-- Change the default ssl_ctx (default nil), when true the connection should try
-- to use the ssl to connect with provider.
-- It's just functional before connect to provider, after connection made, this
-- function will have no effect.
--
-- @param ssl_ctx the context ssl
function ctx:set_ssl_ctx(ssl_ctx)
self.ssl_ctx = ssl_ctx
end
-- Set max channels
-- This function enable to set the max channels to this connection.
-- It's just functional before connect to provider, after connection made, this
-- function will have no effect.
--
-- @param channel_max the max channel
function ctx:set_channel_max(channel_max)
self.channel_max = channel_max or consts.DEFAULT_MAX_CHANNELS,
end
-- Set frame max size
-- This function enable to set the max size of frame. Take
-- It's just functional before connect to provider, after connection made, this
-- function will have no effect.
--
-- @param frame_max the frame max size
function ctx:set_frame_max(frame_max)
self.frame_max = frame_max
end
return ctx
end
return new

22
src/amqp.bkp/commons/validate.lua Normal file
View File

@ -0,0 +1,22 @@
local log = require 'amqp.log.selector'
local validate = {}
function validate.opts(opts)
if not opts then
log.err("no opts provided")
os.exit()
end
if type(opts) ~= "table" then
log.err("opts is not valid")
os.exit()
end
if (opts.role == nil or opts.role == "consumer") and not opts.queue then
log.err("The 'queue' is required to connect as a consumer")
os.exit()
end
end
return validate

1049
src/amqp.bkp/init.lua Normal file
View File

File diff suppressed because it is too large Load Diff

1049
src/amqp.bkp/init.lua.bkp Normal file
View File

File diff suppressed because it is too large Load Diff

11
src/amqp.bkp/log/selector.lua Normal file
View File

@ -0,0 +1,11 @@
if _G.ngx and _G.ngx.socket then
-- not implemented yet
end
local use_cqueues, lfs = pcall(require,"cqueues")
if use_cqueues == true then
-- not implemented yet
end
-- select the lua socket and their extensions
return require 'amqp.log.simple'

90
src/amqp.bkp/log/simple.lua Normal file
View File

@ -0,0 +1,90 @@
--
-- log.lua
--
-- Copyright (c) 2016 rxi
--
-- This library is free software; you can redistribute it and/or modify it
-- under the terms of the MIT license. See LICENSE for details.
--
local log = { _version = "0.1.0" }
log.usecolor = true
log.outfile = nil
log.level = "trace"
local modes = {
{ name = "trace", color = "\27[34m", },
{ name = "debug", color = "\27[36m", },
{ name = "info", color = "\27[32m", },
{ name = "warn", color = "\27[33m", },
{ name = "error", color = "\27[31m", },
{ name = "fatal", color = "\27[35m", },
}
local levels = {}
for i, v in ipairs(modes) do
levels[v.name] = i
end
local round = function(x, increment)
increment = increment or 1
x = x / increment
return (x > 0 and math.floor(x + .5) or math.ceil(x - .5)) * increment
end
local _tostring = tostring
local tostring = function(...)
local t = {}
for i = 1, select('#', ...) do
local x = select(i, ...)
if type(x) == "number" then
x = round(x, .01)
end
t[#t + 1] = _tostring(x)
end
return table.concat(t, " ")
end
for i, x in ipairs(modes) do
local nameupper = x.name:upper()
log[x.name] = function(...)
-- Return early if we're below the log level
if i < levels[log.level] then
return
end
local msg = tostring(...)
local info = debug.getinfo(2, "Sl")
local lineinfo = info.short_src .. ":" .. info.currentline
-- Output to console
print(string.format("%s[%-6s%s]%s %s: %s",
log.usecolor and x.color or "",
nameupper,
os.date("%H:%M:%S"),
log.usecolor and "\27[0m" or "",
lineinfo,
msg))
-- Output to log file
if log.outfile then
local fp = io.open(log.outfile, "a")
local str = string.format("[%-6s%s] %s: %s\n",
nameupper, os.date(), lineinfo, msg)
fp:write(str)
fp:close()
end
end
end
return log

95
src/amqp.bkp/logger.lua Normal file
View File

@ -0,0 +1,95 @@
--
-- Copyright (C) 2016 Meng Zhang @ Yottaa,Inc
-- Copyright (C) 2018 4mig4
--
-- logger
--
local logger = {}
-- logging scaffold
local log
if _G.ngx and _G.ngx.log then
log = _G.ngx.log
else
log = print
end
-- logging level for print
local ERR = 4
local INFO = 7
local DEBUG = 8
-- ngx.log requires a number to indicate the logging level
if _G.ngx then
ERR = _G.ngx.ERR
INFO = _G.ngx.INFO
DEBUG = _G.ngx.DEBUG
end
local level_ = INFO
local function to_string(v_)
if v_ == nil then
return ""
end
if type(v_) ~= "table" then
return tostring(v_)
end
local s = "["
for k,v in pairs(v_) do
if k ~= nil then
s = s .. to_string(k) .. ":"
end
if v ~= nil then
s = s .. to_string(v)
end
s = s .. " "
end
s = s .. "]"
return s
end
local function va_table_to_string(tbl)
local res = ""
for _,v in pairs(tbl) do
res = res .. to_string(v) .. "\t"
end
return res
end
function logger.set_level(level)
level_ = level
end
function logger.error(...)
if level_ < ERR then
return
end
log(ERR, va_table_to_string({...}))
end
function logger.info(...)
if level_ < INFO then
return
end
log(INFO, va_table_to_string({...}))
end
function logger.dbg(...)
if level_ ~= DEBUG then
return
end
log(DEBUG, va_table_to_string({...}))
end
function logger.is_debug_enabled()
return level_ == DEBUG
end
return logger

11
src/amqp/amqp.lua Normal file
View File

@ -0,0 +1,11 @@
local consts = require ('amqp.consts')
local frame = require ('amqp.frame')
local socket = require('amqp.socket.selector')
local bit = require('bit')
local amqp = {}

386
src/amqp/buffer.lua Normal file
View File

@ -0,0 +1,386 @@
--
-- Copyright (C) 2016 Meng Zhang @ Yottaa,Inc
-- Copyright (C) 2018 4mig4
--
local logger = require('amqp.logger')
local bit = require('bit')
local band = bit.band
local bor = bit.bor
local lshift = bit.lshift
local rshift = bit.rshift
local tohex = bit.tohex
local concat = table.concat
local sub = string.sub
local byte = string.byte
local char = string.char
local format = string.format
local ok, new_tab = pcall(require, "table.new")
if not ok or type(new_tab) ~= "function" then
new_tab = function (_,_) return {} end
end
local buffer = {}
local mt = { __index = buffer }
function buffer.new(b,pos)
return setmetatable( { buffer_ = b or "", pos_ = pos or 1 }, mt)
end
function buffer:hex_dump()
local len = #self.buffer_
local bytes = new_tab(len, 0)
for i = 1, len do
bytes[i] = tohex(byte(self.buffer_, i), 2)
end
return concat(bytes, " ")
end
function buffer:get_i8()
local b = byte(self.buffer_, self.pos_)
self.pos_ = self.pos_ + 1
return b
end
function buffer:get_bool()
local v = self:get_i8()
return v and v ~= 0
end
function buffer:get_i16()
local a0, a1 = byte(self.buffer_, self.pos_, self.pos_ + 1)
local r = bor(a1, lshift(a0, 8))
self.pos_ = self.pos_ + 2
return r
end
function buffer:get_i24()
local a0, a1, a2 = byte(self.buffer_, self.pos_, self.pos_ + 2)
self.pos_ = self.pos_ + 3
return bor(a2,
lshift(a1, 8),
lshift(a0, 16))
end
function buffer:get_i32()
local a0, a1, a2, a3 = byte(self.buffer_, self.pos_, self.pos_ + 3)
self.pos_ = self.pos_ + 4
return bor(a3,
lshift(a2, 8),
lshift(a1, 16),
lshift(a0, 24))
end
function buffer:get_i64()
local a, b, c, d, e, f, g, h = byte(self.buffer_, self.pos_, self.pos_ + 7)
self.pos_ = self.pos_ + 8
local lo = bor(h, lshift(g, 8), lshift(f, 16), lshift(e, 24))
local hi = bor(d, lshift(c, 8), lshift(b, 16), lshift(a, 24))
return lo + hi * 4294967296
end
function buffer:get_short_string()
local length = self:get_i8()
local tail = self.pos_+length-1
local s = sub(self.buffer_,self.pos_,tail)
self.pos_ = tail + 1
return s
end
function buffer:get_long_string()
local length = self:get_i32()
local tail = self.pos_+length-1
local s = sub(self.buffer_,self.pos_,tail)
self.pos_ = tail + 1
return s
end
function buffer:get_decimal()
local scale = self:get_i8()
local value = self:get_i32()
local d = {scale = scale, value = value}
return d
end
function buffer:get_f32()
return self:get_i32()
end
function buffer:get_f64()
return self:get_i64()
end
function buffer:get_timestamp()
return self:get_i64()
end
function buffer:get_field_array()
local size = self:get_i32()
logger.dbg("[array] size: " .. size)
local a = {}
local p = self.pos_
while size > 0 do
local f = self:field_value()
a[#a+1] = f
size = size - (self.pos_ - p)
p = self.pos_
end
return a
end
function buffer:get_field_table()
local size = self:get_i32()
local r = {}
local p = self.pos_
local k,v
while size > 0 do
k = self:get_short_string()
v = self:field_value()
size = size - self.pos_ + p
p = self.pos_
r[k] = v
end
return r
end
function buffer:put_i8(i)
self.buffer_ = self.buffer_ .. char(band(i,0x0ff))
end
function buffer:put_bool(b)
local v = 0
if b and b ~= 0 then
v = 1
end
self:put_i8(v)
end
function buffer:put_i16(i)
self.buffer_ = self.buffer_ ..
char(rshift(band(i,0xff00),8)) ..
char(band(i,0x0ff))
end
function buffer:put_i32(i)
self.buffer_ = self.buffer_ ..
char(rshift(band(i,0xff000000),24)) ..
char(rshift(band(i, 0x00ff0000),16)) ..
char(rshift(band(i, 0x0000ff00),8)) ..
char(band(i, 0x000000ff))
end
function buffer:put_i64(i)
-- rshift has not support for 64bit?
-- side effect is that it will rotate for shifts bigger than 32bit
local hi = band(i/4294967296,0x0ffffffff)
local lo = band(i, 0x0ffffffff)
self:put_i32(hi)
self:put_i32(lo)
end
function buffer:put_f32(i)
self:put_i32(i)
end
function buffer:put_f64(i)
self:put_i64(i)
end
function buffer:put_timestamp(i)
self:put_i64(i)
end
function buffer:put_decimal(d)
self:put_i8(d.scale)
self:put_i32(d.value)
end
function buffer:put_short_string(s)
local len = #s
self:put_i8(len)
self.buffer_ = self.buffer_ .. s
end
function buffer:put_long_string(s)
local len = #s
self:put_i32(len)
self.buffer_ = self.buffer_ .. s
end
function buffer:put_payload(payload)
self.buffer_ = self.buffer_ .. payload
end
local function is_array(t)
local i = 0
for _ in pairs(t) do
i = i + 1
if t[i] == nil then return false end
end
return true
end
function buffer:put_field_array(a)
local b = buffer.new()
for i = 1, #a do
b:put_field_value(a[i])
end
self:put_i32(#b.buffer_)
self:put_payload(b.buffer_)
end
function buffer:put_field_table(tab)
local b = buffer.new()
for k,v in pairs(tab) do
b:put_short_string(k)
b:put_field_value(v)
end
self:put_i32(#b.buffer_)
self:put_payload(b.buffer_)
end
local fields_ = {
t = {
r = function (self)
local b = self:get_i8()
return b ~= 0
end,
w = function (self,val)
local b = 0
if val ~= 0 then
b = 1
end
self:put_i8(b)
end
},
b = {
r = buffer.get_i8,
w = buffer.put_i8
},
B = {
r = buffer.get_i8,
w = buffer.put_i8
},
U = {
r = buffer.get_i16,
w = buffer.put_i16
},
u = {
r = buffer.get_i16,
w = buffer.put_i16
},
I = {
r = buffer.get_i32,
w = buffer.put_i32
},
i = {
r = buffer.get_i32,
w = buffer.put_i32
},
L = {
r = buffer.get_i64,
w = buffer.put_i64
},
l = {
r = buffer.get_i64,
w = buffer.put_i64
},
f = {
r = buffer.get_f32,
w = buffer.put_f32
},
d = {
r = buffer.get_f64,
w = buffer.put_f64
},
D = {
r = buffer.get_decimal,
w = buffer.put_decimal
},
s = {
r = buffer.get_short_string,
w = buffer.put_short_string
},
S = {
r = buffer.get_long_string,
w = buffer.put_long_string
},
A = {
r = buffer.get_field_array,
w = buffer.put_field_array
},
T = {
r = buffer.get_timestamp,
w = buffer.put_timestamp
},
F = {
r = buffer.get_field_table,
w = buffer.put_field_table
},
x = {
r = buffer.get_long_string,
w = buffer.put_long_string
},
V = {
r = function()
return nil
end,
w = nil
}
}
function buffer:field_value()
local typ = self:get_i8()
local codec = fields_[char(typ)]
if not codec then
local err = format("codec[%d] not found.",typ)
logger.error("[field_value] " .. err)
return nil,err
end
return codec.r(self)
end
function buffer:put_field_value(value)
-- FIXME: to detect the type of the value
local t = type(value)
local typ = nil
if t == 'number' then
typ = 'I' -- assume to be i32
elseif t == 'boolean' then
typ = 't'
elseif t == 'string' then
typ = 'S'
elseif t == 'table' then
typ = 'F'
if is_array(value) then
typ = 'A'
end
end
-- wire the type
self:put_i8(byte(typ))
local codec = fields_[typ]
if not codec then
local err = format("codec[%d] not found.",typ)
logger.error("[field_value] " .. err)
return nil,err
end
codec.w(self,value)
end
function buffer:payload()
return self.buffer_
end
return buffer

175
src/amqp/consts.lua Normal file
View File

@ -0,0 +1,175 @@
--
-- Copyright (C) 2016 Meng Zhang @ Yottaa,Inc
-- Copyright (C) 2018 @4mig4
-- Copyright (C) 2021 @gsdenys
--
-- Constants definition
--
local amqp = {
DEFAULT_TIMEOUT = 5000, -- 5s
DEFAULT_CHANNEL = 0,
DEFAULT_FRAME_SIZE = 131072,
DEFAULT_MAX_CHANNELS = 65535,
DEFAULT_HEARTBEAT = 30,
PROTOCOL_VERSION_MAJOR = 0,
PROTOCOL_VERSION_MINOR = 9,
PROTOCOL_VERSION_REVISION = 1,
PROTOCOL_PORT = 5672,
PROTOCOL_SSL_PORT = 5671,
state = {
CLOSED = 0,
ESTABLISHED = 1,
CLOSE_WAIT = 2,
},
frame = {
METHOD_FRAME = 1,
HEADER_FRAME = 2,
BODY_FRAME = 3,
HEARTBEAT_FRAME = 8,
FRAME_MIN_SIZE = 4096,
FRAME_END = 0xCE
},
method = {
connection = {
START = 0x0A,
START_OK = 0x0B,
SECURE = 0x14,
SECURE_OK = 0x15,
TUNE = 0x1E,
TUNE_OK = 0x1F,
OPEN = 0x28,
OPEN_OK = 0x29,
CLOSE = 0x32,
CLOSE_OK = 0x33,
BLOCKED = 0x3C,
UNBLOCKED = 0x3D,
},
channel = {
OPEN = 0x0A,
OPEN_OK = 0x0B,
FLOW = 0x14,
FLOW_OK = 0x15,
CLOSE = 0x28,
CLOSE_OK = 0x29,
},
exchange = {
DECLARE = 0x0A,
DECLARE_OK = 0x0B,
DELETE = 0x14,
DELETE_OK = 0x15,
BIND = 0x1E,
BIND_OK = 0x1F,
UNBIND = 0x28,
UNBIND_OK = 0x33,
},
queue = {
DECLARE = 0x0A,
DECLARE_OK = 0x0B,
BIND = 0x14,
BIND_OK = 0x15,
PURGE = 0x1E,
PURGE_OK = 0x1F,
DELETE = 0x28,
DELETE_OK = 0x29,
UNBIND = 0x32,
UNBIND_OK = 0x33,
},
basic = {
QOS = 0x0A,
QOS_OK = 0x0B,
CONSUME = 0x14,
CONSUME_OK = 0x15,
CANCEL = 0x1E,
CANCEL_OK = 0x1F,
PUBLISH = 0x28,
RETURN = 0x32,
DELIVER = 0x3C,
GET = 0x46,
GET_OK = 0x47,
GET_EMPTY = 0x48,
ACK = 0x50,
REJECT = 0x5A,
RECOVER_ASYNC = 0x64,
RECOVER = 0x6E,
RECOVER_OK = 0x6F,
NACK = 0x78,
},
tx = {
SELECT = 0x0A,
SELECT_OK = 0x0B,
COMMIT = 0x14,
COMMIT_OK = 0x15,
ROLLBACK = 0x1E,
ROLLBACK_OK = 0x1F,
},
confirm = {
SELECT = 0x0A,
SELECT_OK = 0x0B
}
},
class = {
CONNECTION = 0x000A,
CHANNEL = 0x0014,
EXCHANGE = 0x0028,
QUEUE = 0x0032,
BASIC = 0x003C,
TX = 0x005A,
CONFIRM = 0x0055
},
flag = {
CONTENT_TYPE = 0x8000,
CONTENT_ENCODING = 0x4000,
HEADERS = 0x2000,
DELIVERY_MODE = 0x1000,
PRIORITY = 0x0800,
CORRELATION_ID = 0x0400,
REPLY_TO = 0x0200,
EXPIRATION = 0x0100,
MESSAGE_ID = 0x0080,
TIMESTAMP = 0x0040,
TYPE = 0x0020,
USER_ID = 0x0010,
APP_ID = 0x0008,
RESERVED1 = 0x0004
},
err = {
REPLY_SUCCESS = 200,
CONTENT_TOO_LARGE = 311,
NO_ROUTE = 312,
NO_CONSUMERS = 313,
CONNECTION_FORCED = 320,
INVALID_PATH = 402,
ACCESS_REFUSED = 403,
NOT_FOUND = 404,
RESOURCE_LOCKED = 405,
PRECONDITION_FAILED = 406,
FRAME_ERROR = 501,
SYNTAX_ERROR = 502,
COMMAND_INVALID = 503,
CHANNEL_ERROR = 504,
UNEXPECTED_FRAME = 505,
RESOURCE_ERROR = 506,
NOT_ALLOWED = 530,
NOT_IMPLEMENTED = 540,
INTERNAL_ERROR = 541
},
PRODUCT = "amqp-client",
VERSION = "2.0.0",
COPYRIGHT = "Copyright (c) 2021 @Yottaa, @4mig4, @gsdenys",
LOCALE = "en_US",
MECHANISM_PLAIN = "PLAIN"
}
return amqp

1590
src/amqp/frame.lua Normal file
View File

File diff suppressed because it is too large Load Diff

55
src/amqp/socket/lua.lua Normal file
View File

@ -0,0 +1,55 @@
local log = require 'amqp.log.simple'
local config = require 'amqp.socket.config'
local basic = {}
basic.socket = require("socket")
basic.tcp = basic.socket.tcp
basic.sock = basic.tcp()
function basic:connect(...)
self.sock:settimeout(config.DEFAULT_TIMEOUT)
local ok, err = self.sock:connect(...)
if not ok then
log.fatal("Error to connect to provider")
os.exit()
end
end
-- Sends message through socket
-- Uses created socket to send AMQP message. This function looks simple
-- but it is necessary to create a patronizing way to call the send
-- information in the differents type of sockets
--
-- @param str message to be sent
function basic:send(str)
return self.sock:send(str)
end
-- Receives message through socket
-- Uses created socket to send AMQP message. This function looks simple
-- but it is necessary to create a patronizing way to call the send
-- information in the differents type of sockets
--
-- @param int the size
function basic:receive(int)
return self.sock:receive(int)
end
-- Close socket connection
-- Enables to close socket connection
function amqp:close()
if not self.sock then
return nil, "not initialized"
end
return self.sock:close()
end
return basic

11
src/amqp/socket/selector.lua Normal file
View File

@ -0,0 +1,11 @@
if _G.ngx and _G.ngx.socket then
-- not implemented yet
end
local use_cqueues, lfs = pcall(require,"cqueues")
if use_cqueues == true then
-- not implemented yet
end
-- select the lua socket and their extensions
return require 'amqp.socket.lua'