diff --git a/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/LICENSE-MPL-RabbitMQ b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/LICENSE-MPL-RabbitMQ new file mode 100644 index 00000000000..4b591e2ba71 --- /dev/null +++ b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/LICENSE-MPL-RabbitMQ @@ -0,0 +1,455 @@ + MOZILLA PUBLIC LICENSE + Version 1.1 + + --------------- + +1. Definitions. + + 1.0.1. "Commercial Use" means distribution or otherwise making the + Covered Code available to a third party. + + 1.1. "Contributor" means each entity that creates or contributes to + the creation of Modifications. + + 1.2. "Contributor Version" means the combination of the Original + Code, prior Modifications used by a Contributor, and the Modifications + made by that particular Contributor. + + 1.3. "Covered Code" means the Original Code or Modifications or the + combination of the Original Code and Modifications, in each case + including portions thereof. + + 1.4. "Electronic Distribution Mechanism" means a mechanism generally + accepted in the software development community for the electronic + transfer of data. + + 1.5. "Executable" means Covered Code in any form other than Source + Code. + + 1.6. "Initial Developer" means the individual or entity identified + as the Initial Developer in the Source Code notice required by Exhibit + A. + + 1.7. "Larger Work" means a work which combines Covered Code or + portions thereof with code not governed by the terms of this License. + + 1.8. "License" means this document. + + 1.8.1. "Licensable" means having the right to grant, to the maximum + extent possible, whether at the time of the initial grant or + subsequently acquired, any and all of the rights conveyed herein. + + 1.9. "Modifications" means any addition to or deletion from the + substance or structure of either the Original Code or any previous + Modifications. When Covered Code is released as a series of files, a + Modification is: + A. Any addition to or deletion from the contents of a file + containing Original Code or previous Modifications. + + B. Any new file that contains any part of the Original Code or + previous Modifications. + + 1.10. "Original Code" means Source Code of computer software code + which is described in the Source Code notice required by Exhibit A as + Original Code, and which, at the time of its release under this + License is not already Covered Code governed by this License. + + 1.10.1. "Patent Claims" means any patent claim(s), now owned or + hereafter acquired, including without limitation, method, process, + and apparatus claims, in any patent Licensable by grantor. + + 1.11. "Source Code" means the preferred form of the Covered Code for + making modifications to it, including all modules it contains, plus + any associated interface definition files, scripts used to control + compilation and installation of an Executable, or source code + differential comparisons against either the Original Code or another + well known, available Covered Code of the Contributor's choice. The + Source Code can be in a compressed or archival form, provided the + appropriate decompression or de-archiving software is widely available + for no charge. + + 1.12. "You" (or "Your") means an individual or a legal entity + exercising rights under, and complying with all of the terms of, this + License or a future version of this License issued under Section 6.1. + For legal entities, "You" includes any entity which controls, is + controlled by, or is under common control with You. For purposes of + this definition, "control" means (a) the power, direct or indirect, + to cause the direction or management of such entity, whether by + contract or otherwise, or (b) ownership of more than fifty percent + (50%) of the outstanding shares or beneficial ownership of such + entity. + +2. Source Code License. + + 2.1. The Initial Developer Grant. + The Initial Developer hereby grants You a world-wide, royalty-free, + non-exclusive license, subject to third party intellectual property + claims: + (a) under intellectual property rights (other than patent or + trademark) Licensable by Initial Developer to use, reproduce, + modify, display, perform, sublicense and distribute the Original + Code (or portions thereof) with or without Modifications, and/or + as part of a Larger Work; and + + (b) under Patents Claims infringed by the making, using or + selling of Original Code, to make, have made, use, practice, + sell, and offer for sale, and/or otherwise dispose of the + Original Code (or portions thereof). + + (c) the licenses granted in this Section 2.1(a) and (b) are + effective on the date Initial Developer first distributes + Original Code under the terms of this License. + + (d) Notwithstanding Section 2.1(b) above, no patent license is + granted: 1) for code that You delete from the Original Code; 2) + separate from the Original Code; or 3) for infringements caused + by: i) the modification of the Original Code or ii) the + combination of the Original Code with other software or devices. + + 2.2. Contributor Grant. + Subject to third party intellectual property claims, each Contributor + hereby grants You a world-wide, royalty-free, non-exclusive license + + (a) under intellectual property rights (other than patent or + trademark) Licensable by Contributor, to use, reproduce, modify, + display, perform, sublicense and distribute the Modifications + created by such Contributor (or portions thereof) either on an + unmodified basis, with other Modifications, as Covered Code + and/or as part of a Larger Work; and + + (b) under Patent Claims infringed by the making, using, or + selling of Modifications made by that Contributor either alone + and/or in combination with its Contributor Version (or portions + of such combination), to make, use, sell, offer for sale, have + made, and/or otherwise dispose of: 1) Modifications made by that + Contributor (or portions thereof); and 2) the combination of + Modifications made by that Contributor with its Contributor + Version (or portions of such combination). + + (c) the licenses granted in Sections 2.2(a) and 2.2(b) are + effective on the date Contributor first makes Commercial Use of + the Covered Code. + + (d) Notwithstanding Section 2.2(b) above, no patent license is + granted: 1) for any code that Contributor has deleted from the + Contributor Version; 2) separate from the Contributor Version; + 3) for infringements caused by: i) third party modifications of + Contributor Version or ii) the combination of Modifications made + by that Contributor with other software (except as part of the + Contributor Version) or other devices; or 4) under Patent Claims + infringed by Covered Code in the absence of Modifications made by + that Contributor. + +3. Distribution Obligations. + + 3.1. Application of License. + The Modifications which You create or to which You contribute are + governed by the terms of this License, including without limitation + Section 2.2. The Source Code version of Covered Code may be + distributed only under the terms of this License or a future version + of this License released under Section 6.1, and You must include a + copy of this License with every copy of the Source Code You + distribute. You may not offer or impose any terms on any Source Code + version that alters or restricts the applicable version of this + License or the recipients' rights hereunder. However, You may include + an additional document offering the additional rights described in + Section 3.5. + + 3.2. Availability of Source Code. + Any Modification which You create or to which You contribute must be + made available in Source Code form under the terms of this License + either on the same media as an Executable version or via an accepted + Electronic Distribution Mechanism to anyone to whom you made an + Executable version available; and if made available via Electronic + Distribution Mechanism, must remain available for at least twelve (12) + months after the date it initially became available, or at least six + (6) months after a subsequent version of that particular Modification + has been made available to such recipients. You are responsible for + ensuring that the Source Code version remains available even if the + Electronic Distribution Mechanism is maintained by a third party. + + 3.3. Description of Modifications. + You must cause all Covered Code to which You contribute to contain a + file documenting the changes You made to create that Covered Code and + the date of any change. You must include a prominent statement that + the Modification is derived, directly or indirectly, from Original + Code provided by the Initial Developer and including the name of the + Initial Developer in (a) the Source Code, and (b) in any notice in an + Executable version or related documentation in which You describe the + origin or ownership of the Covered Code. + + 3.4. Intellectual Property Matters + (a) Third Party Claims. + If Contributor has knowledge that a license under a third party's + intellectual property rights is required to exercise the rights + granted by such Contributor under Sections 2.1 or 2.2, + Contributor must include a text file with the Source Code + distribution titled "LEGAL" which describes the claim and the + party making the claim in sufficient detail that a recipient will + know whom to contact. If Contributor obtains such knowledge after + the Modification is made available as described in Section 3.2, + Contributor shall promptly modify the LEGAL file in all copies + Contributor makes available thereafter and shall take other steps + (such as notifying appropriate mailing lists or newsgroups) + reasonably calculated to inform those who received the Covered + Code that new knowledge has been obtained. + + (b) Contributor APIs. + If Contributor's Modifications include an application programming + interface and Contributor has knowledge of patent licenses which + are reasonably necessary to implement that API, Contributor must + also include this information in the LEGAL file. + + (c) Representations. + Contributor represents that, except as disclosed pursuant to + Section 3.4(a) above, Contributor believes that Contributor's + Modifications are Contributor's original creation(s) and/or + Contributor has sufficient rights to grant the rights conveyed by + this License. + + 3.5. Required Notices. + You must duplicate the notice in Exhibit A in each file of the Source + Code. If it is not possible to put such notice in a particular Source + Code file due to its structure, then You must include such notice in a + location (such as a relevant directory) where a user would be likely + to look for such a notice. If You created one or more Modification(s) + You may add your name as a Contributor to the notice described in + Exhibit A. You must also duplicate this License in any documentation + for the Source Code where You describe recipients' rights or ownership + rights relating to Covered Code. You may choose to offer, and to + charge a fee for, warranty, support, indemnity or liability + obligations to one or more recipients of Covered Code. However, You + may do so only on Your own behalf, and not on behalf of the Initial + Developer or any Contributor. You must make it absolutely clear than + any such warranty, support, indemnity or liability obligation is + offered by You alone, and You hereby agree to indemnify the Initial + Developer and every Contributor for any liability incurred by the + Initial Developer or such Contributor as a result of warranty, + support, indemnity or liability terms You offer. + + 3.6. Distribution of Executable Versions. + You may distribute Covered Code in Executable form only if the + requirements of Section 3.1-3.5 have been met for that Covered Code, + and if You include a notice stating that the Source Code version of + the Covered Code is available under the terms of this License, + including a description of how and where You have fulfilled the + obligations of Section 3.2. The notice must be conspicuously included + in any notice in an Executable version, related documentation or + collateral in which You describe recipients' rights relating to the + Covered Code. You may distribute the Executable version of Covered + Code or ownership rights under a license of Your choice, which may + contain terms different from this License, provided that You are in + compliance with the terms of this License and that the license for the + Executable version does not attempt to limit or alter the recipient's + rights in the Source Code version from the rights set forth in this + License. If You distribute the Executable version under a different + license You must make it absolutely clear that any terms which differ + from this License are offered by You alone, not by the Initial + Developer or any Contributor. You hereby agree to indemnify the + Initial Developer and every Contributor for any liability incurred by + the Initial Developer or such Contributor as a result of any such + terms You offer. + + 3.7. Larger Works. + You may create a Larger Work by combining Covered Code with other code + not governed by the terms of this License and distribute the Larger + Work as a single product. In such a case, You must make sure the + requirements of this License are fulfilled for the Covered Code. + +4. Inability to Comply Due to Statute or Regulation. + + If it is impossible for You to comply with any of the terms of this + License with respect to some or all of the Covered Code due to + statute, judicial order, or regulation then You must: (a) comply with + the terms of this License to the maximum extent possible; and (b) + describe the limitations and the code they affect. Such description + must be included in the LEGAL file described in Section 3.4 and must + be included with all distributions of the Source Code. Except to the + extent prohibited by statute or regulation, such description must be + sufficiently detailed for a recipient of ordinary skill to be able to + understand it. + +5. Application of this License. + + This License applies to code to which the Initial Developer has + attached the notice in Exhibit A and to related Covered Code. + +6. Versions of the License. + + 6.1. New Versions. + Netscape Communications Corporation ("Netscape") may publish revised + and/or new versions of the License from time to time. Each version + will be given a distinguishing version number. + + 6.2. Effect of New Versions. + Once Covered Code has been published under a particular version of the + License, You may always continue to use it under the terms of that + version. You may also choose to use such Covered Code under the terms + of any subsequent version of the License published by Netscape. No one + other than Netscape has the right to modify the terms applicable to + Covered Code created under this License. + + 6.3. Derivative Works. + If You create or use a modified version of this License (which you may + only do in order to apply it to code which is not already Covered Code + governed by this License), You must (a) rename Your license so that + the phrases "Mozilla", "MOZILLAPL", "MOZPL", "Netscape", + "MPL", "NPL" or any confusingly similar phrase do not appear in your + license (except to note that your license differs from this License) + and (b) otherwise make it clear that Your version of the license + contains terms which differ from the Mozilla Public License and + Netscape Public License. (Filling in the name of the Initial + Developer, Original Code or Contributor in the notice described in + Exhibit A shall not of themselves be deemed to be modifications of + this License.) + +7. DISCLAIMER OF WARRANTY. + + COVERED CODE IS PROVIDED UNDER THIS LICENSE ON AN "AS IS" BASIS, + WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING, + WITHOUT LIMITATION, WARRANTIES THAT THE COVERED CODE IS FREE OF + DEFECTS, MERCHANTABLE, FIT FOR A PARTICULAR PURPOSE OR NON-INFRINGING. + THE ENTIRE RISK AS TO THE QUALITY AND PERFORMANCE OF THE COVERED CODE + IS WITH YOU. SHOULD ANY COVERED CODE PROVE DEFECTIVE IN ANY RESPECT, + YOU (NOT THE INITIAL DEVELOPER OR ANY OTHER CONTRIBUTOR) ASSUME THE + COST OF ANY NECESSARY SERVICING, REPAIR OR CORRECTION. THIS DISCLAIMER + OF WARRANTY CONSTITUTES AN ESSENTIAL PART OF THIS LICENSE. NO USE OF + ANY COVERED CODE IS AUTHORIZED HEREUNDER EXCEPT UNDER THIS DISCLAIMER. + +8. TERMINATION. + + 8.1. This License and the rights granted hereunder will terminate + automatically if You fail to comply with terms herein and fail to cure + such breach within 30 days of becoming aware of the breach. All + sublicenses to the Covered Code which are properly granted shall + survive any termination of this License. Provisions which, by their + nature, must remain in effect beyond the termination of this License + shall survive. + + 8.2. If You initiate litigation by asserting a patent infringement + claim (excluding declatory judgment actions) against Initial Developer + or a Contributor (the Initial Developer or Contributor against whom + You file such action is referred to as "Participant") alleging that: + + (a) such Participant's Contributor Version directly or indirectly + infringes any patent, then any and all rights granted by such + Participant to You under Sections 2.1 and/or 2.2 of this License + shall, upon 60 days notice from Participant terminate prospectively, + unless if within 60 days after receipt of notice You either: (i) + agree in writing to pay Participant a mutually agreeable reasonable + royalty for Your past and future use of Modifications made by such + Participant, or (ii) withdraw Your litigation claim with respect to + the Contributor Version against such Participant. If within 60 days + of notice, a reasonable royalty and payment arrangement are not + mutually agreed upon in writing by the parties or the litigation claim + is not withdrawn, the rights granted by Participant to You under + Sections 2.1 and/or 2.2 automatically terminate at the expiration of + the 60 day notice period specified above. + + (b) any software, hardware, or device, other than such Participant's + Contributor Version, directly or indirectly infringes any patent, then + any rights granted to You by such Participant under Sections 2.1(b) + and 2.2(b) are revoked effective as of the date You first made, used, + sold, distributed, or had made, Modifications made by that + Participant. + + 8.3. If You assert a patent infringement claim against Participant + alleging that such Participant's Contributor Version directly or + indirectly infringes any patent where such claim is resolved (such as + by license or settlement) prior to the initiation of patent + infringement litigation, then the reasonable value of the licenses + granted by such Participant under Sections 2.1 or 2.2 shall be taken + into account in determining the amount or value of any payment or + license. + + 8.4. In the event of termination under Sections 8.1 or 8.2 above, + all end user license agreements (excluding distributors and resellers) + which have been validly granted by You or any distributor hereunder + prior to termination shall survive termination. + +9. LIMITATION OF LIABILITY. + + UNDER NO CIRCUMSTANCES AND UNDER NO LEGAL THEORY, WHETHER TORT + (INCLUDING NEGLIGENCE), CONTRACT, OR OTHERWISE, SHALL YOU, THE INITIAL + DEVELOPER, ANY OTHER CONTRIBUTOR, OR ANY DISTRIBUTOR OF COVERED CODE, + OR ANY SUPPLIER OF ANY OF SUCH PARTIES, BE LIABLE TO ANY PERSON FOR + ANY INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES OF ANY + CHARACTER INCLUDING, WITHOUT LIMITATION, DAMAGES FOR LOSS OF GOODWILL, + WORK STOPPAGE, COMPUTER FAILURE OR MALFUNCTION, OR ANY AND ALL OTHER + COMMERCIAL DAMAGES OR LOSSES, EVEN IF SUCH PARTY SHALL HAVE BEEN + INFORMED OF THE POSSIBILITY OF SUCH DAMAGES. THIS LIMITATION OF + LIABILITY SHALL NOT APPLY TO LIABILITY FOR DEATH OR PERSONAL INJURY + RESULTING FROM SUCH PARTY'S NEGLIGENCE TO THE EXTENT APPLICABLE LAW + PROHIBITS SUCH LIMITATION. SOME JURISDICTIONS DO NOT ALLOW THE + EXCLUSION OR LIMITATION OF INCIDENTAL OR CONSEQUENTIAL DAMAGES, SO + THIS EXCLUSION AND LIMITATION MAY NOT APPLY TO YOU. + +10. U.S. GOVERNMENT END USERS. + + The Covered Code is a "commercial item," as that term is defined in + 48 C.F.R. 2.101 (Oct. 1995), consisting of "commercial computer + software" and "commercial computer software documentation," as such + terms are used in 48 C.F.R. 12.212 (Sept. 1995). Consistent with 48 + C.F.R. 12.212 and 48 C.F.R. 227.7202-1 through 227.7202-4 (June 1995), + all U.S. Government End Users acquire Covered Code with only those + rights set forth herein. + +11. MISCELLANEOUS. + + This License represents the complete agreement concerning subject + matter hereof. If any provision of this License is held to be + unenforceable, such provision shall be reformed only to the extent + necessary to make it enforceable. This License shall be governed by + California law provisions (except to the extent applicable law, if + any, provides otherwise), excluding its conflict-of-law provisions. + With respect to disputes in which at least one party is a citizen of, + or an entity chartered or registered to do business in the United + States of America, any litigation relating to this License shall be + subject to the jurisdiction of the Federal Courts of the Northern + District of California, with venue lying in Santa Clara County, + California, with the losing party responsible for costs, including + without limitation, court costs and reasonable attorneys' fees and + expenses. The application of the United Nations Convention on + Contracts for the International Sale of Goods is expressly excluded. + Any law or regulation which provides that the language of a contract + shall be construed against the drafter shall not apply to this + License. + +12. RESPONSIBILITY FOR CLAIMS. + + As between Initial Developer and the Contributors, each party is + responsible for claims and damages arising, directly or indirectly, + out of its utilization of rights under this License and You agree to + work with Initial Developer and Contributors to distribute such + responsibility on an equitable basis. Nothing herein is intended or + shall be deemed to constitute any admission of liability. + +13. MULTIPLE-LICENSED CODE. + + Initial Developer may designate portions of the Covered Code as + "Multiple-Licensed". "Multiple-Licensed" means that the Initial + Developer permits you to utilize portions of the Covered Code under + Your choice of the NPL or the alternative licenses, if any, specified + by the Initial Developer in the file described in Exhibit A. + +EXHIBIT A -Mozilla Public License. + + ``The contents of this file are subject to the Mozilla Public License + Version 1.1 (the "License"); you may not use this file except in + compliance with the License. You may obtain a copy of the License at + http://www.mozilla.org/MPL/ + + Software distributed under the License is distributed on an "AS IS" + basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the + License for the specific language governing rights and limitations + under the License. + + The Original Code is RabbitMQ Consistent Hash Exchange. + + The Initial Developer of the Original Code is GoPivotal, Inc. + Copyright (c) 2014 GoPivotal, Inc. All rights reserved.'' + + [NOTE: The text of this Exhibit A may differ slightly from the text of + the notices in the Source Code files of the Original Code. You should + use the text of this Exhibit A rather than the text found in the + Original Code Source Code for Your Modifications.] diff --git a/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/Makefile b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/Makefile new file mode 100644 index 00000000000..559ffc8be07 --- /dev/null +++ b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/Makefile @@ -0,0 +1,28 @@ +include ../umbrella.mk + +RABBITMQCTL=../rabbitmq-server/scripts/rabbitmqctl +TEST_TMPDIR=$(TMPDIR)/rabbitmq-test +OTHER_NODE=undefined +OTHER_PORT=undefined + +start-other-node: + rm -f $(TEST_TMPDIR)/rabbitmq-$(OTHER_NODE)-pid + RABBITMQ_MNESIA_BASE=$(TEST_TMPDIR)/rabbitmq-$(OTHER_NODE)-mnesia \ + RABBITMQ_PID_FILE=$(TEST_TMPDIR)/rabbitmq-$(OTHER_NODE)-pid \ + RABBITMQ_LOG_BASE=$(TEST_TMPDIR)/log \ + RABBITMQ_NODENAME=$(OTHER_NODE) \ + RABBITMQ_NODE_PORT=$(OTHER_PORT) \ + RABBITMQ_CONFIG_FILE=etc/$(OTHER_NODE) \ + RABBITMQ_PLUGINS_DIR=$(TEST_TMPDIR)/plugins \ + RABBITMQ_PLUGINS_EXPAND_DIR=$(TEST_TMPDIR)/$(OTHER_NODE)-plugins-expand \ + ../rabbitmq-server/scripts/rabbitmq-server >/tmp/$(OTHER_NODE).out 2>/tmp/$(OTHER_NODE).err & + $(RABBITMQCTL) -n $(OTHER_NODE) wait $(TEST_TMPDIR)/rabbitmq-$(OTHER_NODE)-pid + +cluster-other-node: + $(RABBITMQCTL) -n $(OTHER_NODE) stop_app + $(RABBITMQCTL) -n $(OTHER_NODE) reset + $(RABBITMQCTL) -n $(OTHER_NODE) join_cluster rabbit-test@`hostname -s` + $(RABBITMQCTL) -n $(OTHER_NODE) start_app + +stop-other-node: + $(RABBITMQCTL) -n $(OTHER_NODE) stop diff --git a/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/README.md b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/README.md new file mode 100644 index 00000000000..1a12b5cbc23 --- /dev/null +++ b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/README.md @@ -0,0 +1,100 @@ +# Overview + +This plugin adds support for priority queues to RabbitMQ. + +# Downloading + +You can download a pre-built binary of this plugin from +http://www.rabbitmq.com/community-plugins.html. + +# Building + +You can build and install it like any other plugin (see +[the plugin development guide](http://www.rabbitmq.com/plugin-development.html)). + +# Declaring priority queues + +Once the plugin is enabled, you can declare priority queues using the +`x-max-priority` argument. This argument should be an integer +indicating the maximum priority the queue should support. For example, +using the Java client: + + Channel ch = ...; + Map args = new HashMap(); + args.put("x-max-priority", 10); + ch.queueDeclare("my-priority-queue", true, false, false, args); + +You can then publish prioritised messages using the `priority` field +of basic.properties. Larger numbers indicate higher priority. + +There is a simple example using the Java client in the `examples` directory. + +## Caution + +While this plugin implements priority queues in terms of standard +queues, it does not support converting between a priority queue and a +standard queue, and the on-disc format is somewhat different. This has +the following implications: + +* _It is dangerous to disable the plugin when durable priority queues exist_; + the broker will fail to start again. Remember that on broker upgrade + non-bundled plugins like this one need to be reinstalled. +* It is similarly dangerous to enable the plugin if you have declared + durable queues with an `x-max-priority` argument without it. I have no + idea why you'd do that, since you wouldn't get priority queues, but + it would also lead to broker crashes on startup. +* Priority queues can only be defined by arguments, not policies. Queues can + never change the number of priorities they support. + +## Argument equivalence + +RabbitMQ does not have a way for plugins to validate queue +arguments. Therefore the usual equivalence-checking that happens with +arguments does not happen here: + +* You can declare a queue with `x-max-priority` and then declare it + again without `x-max-priority`; no error will result. +* Conversely, you can declare a queue without `x-max-priority` and then + declare it again with `x-max-priority`; again no error will result, + (but the queue will not become a priority queue). +* You can declare a queue with an `x-max-priority` argument which is not + an integer. The plugin will ignore this argument. + +# Behaviour + +The AMQP spec is a bit vague about how priorities work. It says that +all queues MUST support at least 2 priorities, and MAY support up to +10. It does not define how messages without a priority property are +treated. + +In contrast to the AMQP spec, RabbitMQ queues by default do not +support priorities. When creating priority queues using this plugin, +you can specify as many priority levels as you like. Note that: + +* There is some in-memory and on-disc cost per priority level per + queue, so you may not wish to create huge numbers of levels. +* The message `priority` field is defined as an unsigned byte, so in + practice priorities should be between 0 and 255. + +Messages without a priority property are treated as if their priority were +0. Messages with a priority which is higher than the queue's +maximum are treated as if they were published with the maximum priority. + +## Interaction with other features + +In general priority queues have all the features of standard RabbitMQ +queues: they support persistence, paging, mirroring, and so on. There +are a couple of interactions that should be noted though: + +* Messages which should expire (as at + http://www.rabbitmq.com/ttl.html) will still only expire from the + head of the queue. This means that unlike with normal queues, even + per-queue TTL can lead to expired lower-priority messages getting + stuck behind non-expired higher priority ones. These messages will + never be delivered, but they will appear in queue statistics. + +* Queues which have a max-length set (as at + http://www.rabbitmq.com/maxlength.html) will, as usual, drop + messages from the head of the queue to enforce the limit. This means + that higher priority messages might be dropped to make way for lower + priority ones, which might not be what you would expect. diff --git a/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/examples/java/src/com/rabbitmq/examples/PriorityQueue.java b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/examples/java/src/com/rabbitmq/examples/PriorityQueue.java new file mode 100644 index 00000000000..5082c49faa0 --- /dev/null +++ b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/examples/java/src/com/rabbitmq/examples/PriorityQueue.java @@ -0,0 +1,51 @@ +package com.rabbitmq.examples; + +import com.rabbitmq.client.AMQP.BasicProperties; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; +import com.rabbitmq.client.DefaultConsumer; +import com.rabbitmq.client.Envelope; +import com.rabbitmq.client.MessageProperties; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; + +public class PriorityQueue { + private static final String QUEUE = "my-priority-queue"; + + public static void main(String[] argv) throws Exception { + ConnectionFactory factory = new ConnectionFactory(); + Connection conn = factory.newConnection(); + Channel ch = conn.createChannel(); + + Map args = new HashMap(); + args.put("x-max-priority", 10); + ch.queueDeclare(QUEUE, true, false, false, args); + + publish(ch, 0); + publish(ch, 5); + publish(ch, 10); + + final CountDownLatch latch = new CountDownLatch(3); + ch.basicConsume(QUEUE, true, new DefaultConsumer(ch) { + @Override + public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { + System.out.println("Received " + new String(body)); + latch.countDown(); + } + }); + + latch.await(); + conn.close(); + } + + private static void publish(Channel ch, int priority) throws IOException { + BasicProperties props = MessageProperties.PERSISTENT_BASIC.builder().priority(priority).build(); + String body = "message with priority " + priority; + System.out.println("Sent " + body); + ch.basicPublish("", QUEUE, props, body.getBytes()); + } +} diff --git a/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/package.mk b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/package.mk new file mode 100644 index 00000000000..29ef16e3a58 --- /dev/null +++ b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/package.mk @@ -0,0 +1,14 @@ +RELEASABLE:=true +DEPS:=rabbitmq-server rabbitmq-erlang-client rabbitmq-test +FILTER:=all +COVER:=false +WITH_BROKER_TEST_COMMANDS:=rabbit_test_runner:run_in_broker(\"$(PACKAGE_DIR)/test/ebin\",\"$(FILTER)\") +STANDALONE_TEST_COMMANDS:=rabbit_test_runner:run_multi(\"$(UMBRELLA_BASE_DIR)/rabbitmq-server\",\"$(PACKAGE_DIR)/test/ebin\",\"$(FILTER)\",$(COVER),\"/tmp/rabbitmq-multi-node/plugins\") + +# NB: we cannot use PACKAGE_DIR in the body of this rule as it gets +# expanded at the wrong time and set to the value of a completely +# arbitrary package! +$(PACKAGE_DIR)+pre-test:: $(PACKAGE_DIR)+dist + rm -rf /tmp/rabbitmq-multi-node/plugins + mkdir -p /tmp/rabbitmq-multi-node/plugins/plugins + cp -p $(UMBRELLA_BASE_DIR)/rabbitmq-priority-queue/dist/*.ez /tmp/rabbitmq-multi-node/plugins/plugins diff --git a/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/src/rabbit_priority_queue.erl b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/src/rabbit_priority_queue.erl new file mode 100644 index 00000000000..ede7953746e --- /dev/null +++ b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/src/rabbit_priority_queue.erl @@ -0,0 +1,539 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License +%% at http://www.mozilla.org/MPL/ +%% +%% Software distributed under the License is distributed on an "AS IS" +%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See +%% the License for the specific language governing rights and +%% limitations under the License. +%% +%% The Original Code is RabbitMQ. +%% +%% The Initial Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2014 GoPivotal, Inc. All rights reserved. +%% + +-module(rabbit_priority_queue). + +-include_lib("rabbit_common/include/rabbit.hrl"). +-include_lib("rabbit_common/include/rabbit_framing.hrl"). +-behaviour(rabbit_backing_queue). + +-rabbit_boot_step({?MODULE, + [{description, "enable priority queue"}, + {mfa, {?MODULE, enable, []}}, + {requires, pre_boot}, + {enables, kernel_ready}]}). + +-export([enable/0]). + +-export([start/1, stop/0]). + +-export([init/3, terminate/2, delete_and_terminate/2, purge/1, purge_acks/1, + publish/5, publish_delivered/4, discard/3, drain_confirmed/1, + dropwhile/2, fetchwhile/4, fetch/2, drop/2, ack/2, requeue/2, + ackfold/4, fold/3, len/1, is_empty/1, depth/1, + set_ram_duration_target/2, ram_duration/1, needs_timeout/1, timeout/1, + handle_pre_hibernate/1, resume/1, msg_rates/1, + status/1, invoke/3, is_duplicate/2]). + +-record(state, {bq, bqss}). +-record(passthrough, {bq, bqs}). + +%% See 'note on suffixes' below +-define(passthrough1(F), State#passthrough{bqs = BQ:F}). +-define(passthrough2(F), + {Res, BQS1} = BQ:F, {Res, State#passthrough{bqs = BQS1}}). +-define(passthrough3(F), + {Res1, Res2, BQS1} = BQ:F, {Res1, Res2, State#passthrough{bqs = BQS1}}). + +enable() -> + {ok, RealBQ} = application:get_env(rabbit, backing_queue_module), + case RealBQ of + ?MODULE -> ok; + _ -> rabbit_log:info("Priority queues enabled, real BQ is ~s~n", + [RealBQ]), + application:set_env( + rabbitmq_priority_queue, backing_queue_module, RealBQ), + application:set_env(rabbit, backing_queue_module, ?MODULE) + end. + +%%---------------------------------------------------------------------------- + +start(QNames) -> + BQ = bq(), + %% TODO this expand-collapse dance is a bit ridiculous but it's what + %% rabbit_amqqueue:recover/0 expects. We could probably simplify + %% this if we rejigged recovery a bit. + {DupNames, ExpNames} = expand_queues(QNames), + case BQ:start(ExpNames) of + {ok, ExpRecovery} -> + {ok, collapse_recovery(QNames, DupNames, ExpRecovery)}; + Else -> + Else + end. + +stop() -> + BQ = bq(), + BQ:stop(). + +%%---------------------------------------------------------------------------- + +mutate_name(P, Q = #amqqueue{name = QName = #resource{name = QNameBin}}) -> + Q#amqqueue{name = QName#resource{name = mutate_name_bin(P, QNameBin)}}. + +mutate_name_bin(P, NameBin) -> <>. + +expand_queues(QNames) -> + lists:unzip( + lists:append([expand_queue(QName) || QName <- QNames])). + +expand_queue(QName = #resource{name = QNameBin}) -> + {ok, Q} = rabbit_misc:dirty_read({rabbit_durable_queue, QName}), + case priorities(Q) of + none -> [{QName, QName}]; + Ps -> [{QName, QName#resource{name = mutate_name_bin(P, QNameBin)}} + || P <- Ps] + end. + +collapse_recovery(QNames, DupNames, Recovery) -> + NameToTerms = lists:foldl(fun({Name, RecTerm}, Dict) -> + dict:append(Name, RecTerm, Dict) + end, dict:new(), lists:zip(DupNames, Recovery)), + [dict:fetch(Name, NameToTerms) || Name <- QNames]. + +priorities(#amqqueue{arguments = Args}) -> + Ints = [long, short, signedint, byte], + case rabbit_misc:table_lookup(Args, <<"x-max-priority">>) of + {Type, Max} -> case lists:member(Type, Ints) of + false -> none; + true -> lists:reverse(lists:seq(0, Max)) + end; + _ -> none + end. + +%%---------------------------------------------------------------------------- + +init(Q, Recover, AsyncCallback) -> + BQ = bq(), + case priorities(Q) of + none -> RealRecover = case Recover of + new -> new; + [R] -> R %% [0] + end, + #passthrough{bq = BQ, + bqs = BQ:init(Q, RealRecover, AsyncCallback)}; + Ps -> Init = fun (P, Term) -> + BQ:init( + mutate_name(P, Q), Term, + fun (M, F) -> AsyncCallback(M, {P, F}) end) + end, + BQSs = case Recover of + new -> [{P, Init(P, new)} || P <- Ps]; + _ -> PsTerms = lists:zip(Ps, Recover), + [{P, Init(P, Term)} || {P, Term} <- PsTerms] + end, + #state{bq = BQ, + bqss = BQSs} + end. +%% [0] collapse_recovery has the effect of making a list of recovery +%% terms in priority order, even for non priority queues. It's easier +%% to do that and "unwrap" in init/3 than to have collapse_recovery be +%% aware of non-priority queues. + +terminate(Reason, State = #state{bq = BQ}) -> + foreach1(fun (_P, BQSN) -> BQ:terminate(Reason, BQSN) end, State); +terminate(Reason, State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough1(terminate(Reason, BQS)). + +delete_and_terminate(Reason, State = #state{bq = BQ}) -> + foreach1(fun (_P, BQSN) -> + BQ:delete_and_terminate(Reason, BQSN) + end, State); +delete_and_terminate(Reason, State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough1(delete_and_terminate(Reason, BQS)). + +purge(State = #state{bq = BQ}) -> + fold_add2(fun (_P, BQSN) -> BQ:purge(BQSN) end, State); +purge(State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough2(purge(BQS)). + +purge_acks(State = #state{bq = BQ}) -> + foreach1(fun (_P, BQSN) -> BQ:purge_acks(BQSN) end, State); +purge_acks(State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough1(purge_acks(BQS)). + +publish(Msg, MsgProps, IsDelivered, ChPid, State = #state{bq = BQ}) -> + pick1(fun (_P, BQSN) -> + BQ:publish(Msg, MsgProps, IsDelivered, ChPid, BQSN) + end, Msg, State); +publish(Msg, MsgProps, IsDelivered, ChPid, + State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough1(publish(Msg, MsgProps, IsDelivered, ChPid, BQS)). + +publish_delivered(Msg, MsgProps, ChPid, State = #state{bq = BQ}) -> + pick2(fun (P, BQSN) -> + {AckTag, BQSN1} = BQ:publish_delivered( + Msg, MsgProps, ChPid, BQSN), + {{P, AckTag}, BQSN1} + end, Msg, State); +publish_delivered(Msg, MsgProps, ChPid, + State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough2(publish_delivered(Msg, MsgProps, ChPid, BQS)). + +%% TODO this is a hack. The BQ api does not give us enough information +%% here - if we had the Msg we could look at its priority and forward +%% to the appropriate sub-BQ. But we don't so we are stuck. +%% +%% But fortunately VQ ignores discard/3, so we can too, *assuming we +%% are talking to VQ*. discard/3 is used by HA, but that's "above" us +%% (if in use) so we don't break that either, just some hypothetical +%% alternate BQ implementation. +discard(_MsgId, _ChPid, State = #state{}) -> + State; + %% We should have something a bit like this here: + %% pick1(fun (_P, BQSN) -> + %% BQ:discard(MsgId, ChPid, BQSN) + %% end, Msg, State); +discard(MsgId, ChPid, State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough1(discard(MsgId, ChPid, BQS)). + +drain_confirmed(State = #state{bq = BQ}) -> + fold_append2(fun (_P, BQSN) -> BQ:drain_confirmed(BQSN) end, State); +drain_confirmed(State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough2(drain_confirmed(BQS)). + +dropwhile(Pred, State = #state{bq = BQ}) -> + find2(fun (_P, BQSN) -> BQ:dropwhile(Pred, BQSN) end, undefined, State); +dropwhile(Pred, State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough2(dropwhile(Pred, BQS)). + +%% TODO this is a bit nasty. In the one place where fetchwhile/4 is +%% actually used the accumulator is a list of acktags, which of course +%% we need to mutate - so we do that although we are encoding an +%% assumption here. +fetchwhile(Pred, Fun, Acc, State = #state{bq = BQ}) -> + findfold3( + fun (P, BQSN, AccN) -> + {Res, AccN1, BQSN1} = BQ:fetchwhile(Pred, Fun, AccN, BQSN), + {Res, priority_on_acktags(P, AccN1), BQSN1} + end, Acc, undefined, State); +fetchwhile(Pred, Fun, Acc, State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough3(fetchwhile(Pred, Fun, Acc, BQS)). + +fetch(AckRequired, State = #state{bq = BQ}) -> + find2( + fun (P, BQSN) -> + case BQ:fetch(AckRequired, BQSN) of + {empty, BQSN1} -> {empty, BQSN1}; + {{Msg, Del, ATag}, BQSN1} -> {{Msg, Del, {P, ATag}}, BQSN1} + end + end, empty, State); +fetch(AckRequired, State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough2(fetch(AckRequired, BQS)). + +drop(AckRequired, State = #state{bq = BQ}) -> + find2(fun (P, BQSN) -> + case BQ:drop(AckRequired, BQSN) of + {empty, BQSN1} -> {empty, BQSN1}; + {{MsgId, AckTag}, BQSN1} -> {{MsgId, {P, AckTag}}, BQSN1} + end + end, empty, State); +drop(AckRequired, State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough2(drop(AckRequired, BQS)). + +ack(AckTags, State = #state{bq = BQ}) -> + fold_by_acktags2(fun (AckTagsN, BQSN) -> + BQ:ack(AckTagsN, BQSN) + end, AckTags, State); +ack(AckTags, State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough2(ack(AckTags, BQS)). + +requeue(AckTags, State = #state{bq = BQ}) -> + fold_by_acktags2(fun (AckTagsN, BQSN) -> + BQ:requeue(AckTagsN, BQSN) + end, AckTags, State); +requeue(AckTags, State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough2(requeue(AckTags, BQS)). + +%% Similar problem to fetchwhile/4 +ackfold(MsgFun, Acc, State = #state{bq = BQ}, AckTags) -> + AckTagsByPriority = partition_acktags(AckTags), + fold2( + fun (P, BQSN, AccN) -> + case orddict:find(P, AckTagsByPriority) of + {ok, ATagsN} -> {AccN1, BQSN1} = + BQ:ackfold(MsgFun, AccN, BQSN, ATagsN), + {priority_on_acktags(P, AccN1), BQSN1}; + error -> {AccN, BQSN} + end + end, Acc, State); +ackfold(MsgFun, Acc, State = #passthrough{bq = BQ, bqs = BQS}, AckTags) -> + ?passthrough2(ackfold(MsgFun, Acc, BQS, AckTags)). + +fold(Fun, Acc, State = #state{bq = BQ}) -> + fold2(fun (_P, BQSN, AccN) -> BQ:fold(Fun, AccN, BQSN) end, Acc, State); +fold(Fun, Acc, State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough2(fold(Fun, Acc, BQS)). + +len(#state{bq = BQ, bqss = BQSs}) -> + add0(fun (_P, BQSN) -> BQ:len(BQSN) end, BQSs); +len(#passthrough{bq = BQ, bqs = BQS}) -> + BQ:len(BQS). + +is_empty(#state{bq = BQ, bqss = BQSs}) -> + all0(fun (_P, BQSN) -> BQ:is_empty(BQSN) end, BQSs); +is_empty(#passthrough{bq = BQ, bqs = BQS}) -> + BQ:is_empty(BQS). + +depth(#state{bq = BQ, bqss = BQSs}) -> + add0(fun (_P, BQSN) -> BQ:depth(BQSN) end, BQSs); +depth(#passthrough{bq = BQ, bqs = BQS}) -> + BQ:depth(BQS). + +set_ram_duration_target(DurationTarget, State = #state{bq = BQ}) -> + foreach1(fun (_P, BQSN) -> + BQ:set_ram_duration_target(DurationTarget, BQSN) + end, State); +set_ram_duration_target(DurationTarget, + State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough1(set_ram_duration_target(DurationTarget, BQS)). + +ram_duration(State = #state{bq = BQ}) -> + fold_add2(fun (_P, BQSN) -> BQ:ram_duration(BQSN) end, State); +ram_duration(State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough2(ram_duration(BQS)). + +needs_timeout(#state{bq = BQ, bqss = BQSs}) -> + fold0(fun (_P, _BQSN, timed) -> timed; + (_P, BQSN, idle) -> case BQ:needs_timeout(BQSN) of + timed -> timed; + _ -> idle + end; + (_P, BQSN, false) -> BQ:needs_timeout(BQSN) + end, false, BQSs); +needs_timeout(#passthrough{bq = BQ, bqs = BQS}) -> + BQ:needs_timeout(BQS). + +timeout(State = #state{bq = BQ}) -> + foreach1(fun (_P, BQSN) -> BQ:timeout(BQSN) end, State); +timeout(State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough1(timeout(BQS)). + +handle_pre_hibernate(State = #state{bq = BQ}) -> + foreach1(fun (_P, BQSN) -> + BQ:handle_pre_hibernate(BQSN) + end, State); +handle_pre_hibernate(State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough1(handle_pre_hibernate(BQS)). + +resume(State = #state{bq = BQ}) -> + foreach1(fun (_P, BQSN) -> BQ:resume(BQSN) end, State); +resume(State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough1(resume(BQS)). + +msg_rates(#state{bq = BQ, bqss = BQSs}) -> + fold0(fun(_P, BQSN, {InN, OutN}) -> + {In, Out} = BQ:msg_rates(BQSN), + {InN + In, OutN + Out} + end, {0.0, 0.0}, BQSs); +msg_rates(#passthrough{bq = BQ, bqs = BQS}) -> + BQ:msg_rates(BQS). + +status(#state{bq = BQ, bqss = BQSs}) -> + fold0(fun (P, BQSN, Acc) -> + combine_status(P, BQ:status(BQSN), Acc) + end, nothing, BQSs); +status(#passthrough{bq = BQ, bqs = BQS}) -> + BQ:status(BQS). + +invoke(Mod, {P, Fun}, State = #state{bq = BQ}) -> + pick1(fun (_P, BQSN) -> BQ:invoke(Mod, Fun, BQSN) end, P, State); +invoke(Mod, Fun, State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough1(invoke(Mod, Fun, BQS)). + +is_duplicate(Msg, State = #state{bq = BQ}) -> + pick2(fun (_P, BQSN) -> BQ:is_duplicate(Msg, BQSN) end, Msg, State); +is_duplicate(Msg, State = #passthrough{bq = BQ, bqs = BQS}) -> + ?passthrough2(is_duplicate(Msg, BQS)). + +%%---------------------------------------------------------------------------- + +bq() -> + {ok, RealBQ} = application:get_env( + rabbitmq_priority_queue, backing_queue_module), + RealBQ. + +%% Note on suffixes: Many utility functions here have suffixes telling +%% you the arity of the return type of the BQ function they are +%% designed to work with. +%% +%% 0 - BQ function returns a value and does not modify state +%% 1 - BQ function just returns a new state +%% 2 - BQ function returns a 2-tuple of {Result, NewState} +%% 3 - BQ function returns a 3-tuple of {Result1, Result2, NewState} + +%% Fold over results +fold0(Fun, Acc, [{P, BQSN} | Rest]) -> fold0(Fun, Fun(P, BQSN, Acc), Rest); +fold0(_Fun, Acc, []) -> Acc. + +%% Do all BQs match? +all0(Pred, BQSs) -> fold0(fun (_P, _BQSN, false) -> false; + (P, BQSN, true) -> Pred(P, BQSN) + end, true, BQSs). + +%% Sum results +add0(Fun, BQSs) -> fold0(fun (P, BQSN, Acc) -> Acc + Fun(P, BQSN) end, 0, BQSs). + +%% Apply for all states +foreach1(Fun, State = #state{bqss = BQSs}) -> + a(State#state{bqss = foreach1(Fun, BQSs, [])}). +foreach1(Fun, [{P, BQSN} | Rest], BQSAcc) -> + BQSN1 = Fun(P, BQSN), + foreach1(Fun, Rest, [{P, BQSN1} | BQSAcc]); +foreach1(_Fun, [], BQSAcc) -> + lists:reverse(BQSAcc). + +%% For a given thing, just go to its BQ +pick1(Fun, Prioritisable, #state{bqss = BQSs} = State) -> + {P, BQSN} = priority(Prioritisable, BQSs), + a(State#state{bqss = bq_store(P, Fun(P, BQSN), BQSs)}). + +%% Fold over results +fold2(Fun, Acc, State = #state{bqss = BQSs}) -> + {Res, BQSs1} = fold2(Fun, Acc, BQSs, []), + {Res, a(State#state{bqss = BQSs1})}. +fold2(Fun, Acc, [{P, BQSN} | Rest], BQSAcc) -> + {Acc1, BQSN1} = Fun(P, BQSN, Acc), + fold2(Fun, Acc1, Rest, [{P, BQSN1} | BQSAcc]); +fold2(_Fun, Acc, [], BQSAcc) -> + {Acc, lists:reverse(BQSAcc)}. + +%% Fold over results assuming results are lists and we want to append them +fold_append2(Fun, State) -> + fold2(fun (P, BQSN, Acc) -> + {Res, BQSN1} = Fun(P, BQSN), + {Res ++ Acc, BQSN1} + end, [], State). + +%% Fold over results assuming results are numbers and we want to sum them +fold_add2(Fun, State) -> + fold2(fun (P, BQSN, Acc) -> + {Res, BQSN1} = Fun(P, BQSN), + {add_maybe_infinity(Res, Acc), BQSN1} + end, 0, State). + +%% Fold over results assuming results are lists and we want to append +%% them, and also that we have some AckTags we want to pass in to each +%% invocation. +fold_by_acktags2(Fun, AckTags, State) -> + AckTagsByPriority = partition_acktags(AckTags), + fold_append2(fun (P, BQSN) -> + case orddict:find(P, AckTagsByPriority) of + {ok, AckTagsN} -> Fun(AckTagsN, BQSN); + error -> {[], BQSN} + end + end, State). + +%% For a given thing, just go to its BQ +pick2(Fun, Prioritisable, #state{bqss = BQSs} = State) -> + {P, BQSN} = priority(Prioritisable, BQSs), + {Res, BQSN1} = Fun(P, BQSN), + {Res, a(State#state{bqss = bq_store(P, BQSN1, BQSs)})}. + +%% Run through BQs in priority order until one does not return +%% {NotFound, NewState} or we have gone through them all. +find2(Fun, NotFound, State = #state{bqss = BQSs}) -> + {Res, BQSs1} = find2(Fun, NotFound, BQSs, []), + {Res, a(State#state{bqss = BQSs1})}. +find2(Fun, NotFound, [{P, BQSN} | Rest], BQSAcc) -> + case Fun(P, BQSN) of + {NotFound, BQSN1} -> find2(Fun, NotFound, Rest, [{P, BQSN1} | BQSAcc]); + {Res, BQSN1} -> {Res, lists:reverse([{P, BQSN1} | BQSAcc]) ++ Rest} + end; +find2(_Fun, NotFound, [], BQSAcc) -> + {NotFound, lists:reverse(BQSAcc)}. + +%% Run through BQs in priority order like find2 but also folding as we go. +findfold3(Fun, Acc, NotFound, State = #state{bqss = BQSs}) -> + {Res, Acc1, BQSs1} = findfold3(Fun, Acc, NotFound, BQSs, []), + {Res, Acc1, a(State#state{bqss = BQSs1})}. +findfold3(Fun, Acc, NotFound, [{P, BQSN} | Rest], BQSAcc) -> + case Fun(P, BQSN, Acc) of + {NotFound, Acc1, BQSN1} -> + findfold3(Fun, Acc1, NotFound, Rest, [{P, BQSN1} | BQSAcc]); + {Res, Acc1, BQSN1} -> + {Res, Acc1, lists:reverse([{P, BQSN1} | BQSAcc]) ++ Rest} + end; +findfold3(_Fun, Acc, NotFound, [], BQSAcc) -> + {NotFound, Acc, lists:reverse(BQSAcc)}. + +bq_fetch(P, []) -> exit({not_found, P}); +bq_fetch(P, [{P, BQSN} | _]) -> BQSN; +bq_fetch(P, [{_, _BQSN} | T]) -> bq_fetch(P, T). + +bq_store(P, BQS, BQSs) -> + [{PN, case PN of + P -> BQS; + _ -> BQSN + end} || {PN, BQSN} <- BQSs]. + +a(State = #state{bqss = BQSs}) -> + Ps = [P || {P, _} <- BQSs], + case lists:reverse(lists:usort(Ps)) of + Ps -> State; + _ -> exit({bad_order, Ps}) + end. + +%%---------------------------------------------------------------------------- + +priority(P, BQSs) when is_integer(P) -> + {P, bq_fetch(P, BQSs)}; +priority(_Msg, [{P, BQSN}]) -> + {P, BQSN}; +priority(Msg = #basic_message{content = #content{properties = Props}}, + [{P, BQSN} | Rest]) -> + #'P_basic'{priority = Priority0} = Props, + Priority = case Priority0 of + undefined -> 0; + _ when is_integer(Priority0) -> Priority0 + end, + case Priority >= P of + true -> {P, BQSN}; + false -> priority(Msg, Rest) + end. + +add_maybe_infinity(infinity, _) -> infinity; +add_maybe_infinity(_, infinity) -> infinity; +add_maybe_infinity(A, B) -> A + B. + +partition_acktags(AckTags) -> partition_acktags(AckTags, orddict:new()). + +partition_acktags([], Partitioned) -> + Partitioned; +partition_acktags([{P, AckTag} | Rest], Partitioned) -> + partition_acktags(Rest, orddict:append(P, AckTag, Partitioned)). + +priority_on_acktags(P, AckTags) -> + [case Tag of + _ when is_integer(Tag) -> {P, Tag}; + _ -> Tag + end || Tag <- AckTags]. + +combine_status(P, New, nothing) -> + [{priorities, [{P, simplify_status(New)}]} | New]; +combine_status(P, New, Old) -> + Combined = [{K, cse(V, proplists:get_value(K, Old))} || {K, V} <- New], + Ps = [{P, simplify_status(New)} | proplists:get_value(priorities, Old)], + [{priorities, Ps} | Combined]. + +simplify_status(Status) -> + [{K, V} || {K, V} <- Status, + lists:member(K, [len, persistent_count, ram_msg_count])]. + +cse(infinity, _) -> infinity; +cse(_, infinity) -> infinity; +cse(A, B) when is_number(A) -> A + B; +cse({delta, _, _, _}, _) -> {delta, todo, todo, todo}; +cse(A, B) -> exit({A, B}). diff --git a/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/src/rabbitmq_priority_queue.app.src b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/src/rabbitmq_priority_queue.app.src new file mode 100644 index 00000000000..d028e9718b9 --- /dev/null +++ b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/src/rabbitmq_priority_queue.app.src @@ -0,0 +1,7 @@ +{application, rabbitmq_priority_queue, + [{description, "Priority queue implementation"}, + {vsn, "%%VSN%%"}, + {modules, []}, + {registered, []}, + {env, []}, + {applications, [rabbit]}]}. diff --git a/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/test/src/rabbit_priority_queue_test.erl b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/test/src/rabbit_priority_queue_test.erl new file mode 100644 index 00000000000..e348db39dac --- /dev/null +++ b/deps/rabbitmq_server-3.3.5/plugins-src/rabbitmq-priority-queue/test/src/rabbit_priority_queue_test.erl @@ -0,0 +1,335 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License +%% at http://www.mozilla.org/MPL/ +%% +%% Software distributed under the License is distributed on an "AS IS" +%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See +%% the License for the specific language governing rights and +%% limitations under the License. +%% +%% The Original Code is RabbitMQ. +%% +%% The Initial Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2014 GoPivotal, Inc. All rights reserved. +%% + +-module(rabbit_priority_queue_test). + +-compile(export_all). +-include_lib("eunit/include/eunit.hrl"). +-include_lib("amqp_client/include/amqp_client.hrl"). + +-import(rabbit_misc, [pget/2]). + +%% The BQ API is used in all sorts of places in all sorts of +%% ways. Therefore we have to jump through a few different hoops +%% in order to integration-test it. +%% +%% * start/1, stop/0, init/3, terminate/2, delete_and_terminate/2 +%% - starting and stopping rabbit. durable queues / persistent msgs needed +%% to test recovery +%% +%% * publish/5, drain_confirmed/1, fetch/2, ack/2, is_duplicate/2, msg_rates/1, +%% needs_timeout/1, timeout/1, invoke/3, resume/1 [0] +%% - regular publishing and consuming, with confirms and acks and durability +%% +%% * publish_delivered/4 - publish with acks straight through +%% * discard/3 - publish without acks straight through +%% * dropwhile/2 - expire messages without DLX +%% * fetchwhile/4 - expire messages with DLX +%% * ackfold/4 - reject messages with DLX +%% * requeue/2 - reject messages without DLX +%% * drop/2 - maxlen messages without DLX +%% * purge/1 - issue AMQP queue.purge +%% * purge_acks/1 - mirror queue explicit sync with unacked msgs +%% * fold/3 - mirror queue explicit sync +%% * depth/1 - mirror queue implicit sync detection +%% * len/1, is_empty/1 - info items +%% * handle_pre_hibernate/1 - hibernation +%% +%% * set_ram_duration_target/2, ram_duration/1, status/1 +%% - maybe need unit testing? +%% +%% [0] publish enough to get credit flow from msg store + +recovery_test() -> + {Conn, Ch} = open(), + Q = <<"test">>, + declare(Ch, Q, 3), + publish(Ch, Q, [1, 2, 3, 1, 2, 3, 1, 2, 3]), + amqp_connection:close(Conn), + + %% TODO these break coverage + rabbit:stop(), + rabbit:start(), + + {Conn2, Ch2} = open(), + get_all(Ch2, Q, do_ack, [3, 3, 3, 2, 2, 2, 1, 1, 1]), + delete(Ch2, Q), + amqp_connection:close(Conn2), + passed. + +simple_order_test() -> + {Conn, Ch} = open(), + Q = <<"test">>, + declare(Ch, Q, 3), + publish(Ch, Q, [1, 2, 3, 1, 2, 3, 1, 2, 3]), + get_all(Ch, Q, do_ack, [3, 3, 3, 2, 2, 2, 1, 1, 1]), + publish(Ch, Q, [2, 3, 1, 2, 3, 1, 2, 3, 1]), + get_all(Ch, Q, no_ack, [3, 3, 3, 2, 2, 2, 1, 1, 1]), + publish(Ch, Q, [3, 1, 2, 3, 1, 2, 3, 1, 2]), + get_all(Ch, Q, do_ack, [3, 3, 3, 2, 2, 2, 1, 1, 1]), + delete(Ch, Q), + amqp_connection:close(Conn), + passed. + +matching_test() -> + {Conn, Ch} = open(), + Q = <<"test">>, + declare(Ch, Q, 5), + %% We round priority down, and 0 is the default + publish(Ch, Q, [undefined, 0, 5, 10, undefined]), + get_all(Ch, Q, do_ack, [5, 10, undefined, 0, undefined]), + delete(Ch, Q), + amqp_connection:close(Conn), + passed. + +resume_test() -> + {Conn, Ch} = open(), + Q = <<"test">>, + declare(Ch, Q, 5), + amqp_channel:call(Ch, #'confirm.select'{}), + publish_many(Ch, Q, 10000), + amqp_channel:wait_for_confirms(Ch), + amqp_channel:call(Ch, #'queue.purge'{queue = Q}), %% Assert it exists + delete(Ch, Q), + amqp_connection:close(Conn), + passed. + +straight_through_test() -> + {Conn, Ch} = open(), + Q = <<"test">>, + declare(Ch, Q, 3), + [begin + consume(Ch, Q, Ack), + [begin + publish1(Ch, Q, P), + assert_delivered(Ch, Ack, P) + end || P <- [1, 2, 3]], + cancel(Ch) + end || Ack <- [do_ack, no_ack]], + get_empty(Ch, Q), + delete(Ch, Q), + amqp_connection:close(Conn), + passed. + +dropwhile_fetchwhile_test() -> + {Conn, Ch} = open(), + Q = <<"test">>, + [begin + declare(Ch, Q, Args ++ arguments(3)), + publish(Ch, Q, [1, 2, 3, 1, 2, 3, 1, 2, 3]), + timer:sleep(10), + get_empty(Ch, Q), + delete(Ch, Q) + end || + Args <- [[{<<"x-message-ttl">>, long, 1}], + [{<<"x-message-ttl">>, long, 1}, + {<<"x-dead-letter-exchange">>, longstr, <<"amq.fanout">>}] + ]], + amqp_connection:close(Conn), + passed. + +ackfold_test() -> + {Conn, Ch} = open(), + Q = <<"test">>, + Q2 = <<"test2">>, + declare(Ch, Q, + [{<<"x-dead-letter-exchange">>, longstr, <<>>}, + {<<"x-dead-letter-routing-key">>, longstr, Q2} + | arguments(3)]), + declare(Ch, Q2, none), + publish(Ch, Q, [1, 2, 3]), + [_, _, DTag] = get_all(Ch, Q, manual_ack, [3, 2, 1]), + amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DTag, + multiple = true, + requeue = false}), + timer:sleep(100), + get_all(Ch, Q2, do_ack, [3, 2, 1]), + delete(Ch, Q), + delete(Ch, Q2), + amqp_connection:close(Conn), + passed. + +requeue_test() -> + {Conn, Ch} = open(), + Q = <<"test">>, + declare(Ch, Q, 3), + publish(Ch, Q, [1, 2, 3]), + [_, _, DTag] = get_all(Ch, Q, manual_ack, [3, 2, 1]), + amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DTag, + multiple = true, + requeue = true}), + get_all(Ch, Q, do_ack, [3, 2, 1]), + delete(Ch, Q), + amqp_connection:close(Conn), + passed. + +drop_test() -> + {Conn, Ch} = open(), + Q = <<"test">>, + declare(Ch, Q, [{<<"x-max-length">>, long, 4} | arguments(3)]), + publish(Ch, Q, [1, 2, 3, 1, 2, 3, 1, 2, 3]), + %% We drop from the head, so this is according to the "spec" even + %% if not likely to be what the user wants. + get_all(Ch, Q, do_ack, [2, 1, 1, 1]), + delete(Ch, Q), + amqp_connection:close(Conn), + passed. + +purge_test() -> + {Conn, Ch} = open(), + Q = <<"test">>, + declare(Ch, Q, 3), + publish(Ch, Q, [1, 2, 3]), + amqp_channel:call(Ch, #'queue.purge'{queue = Q}), + get_empty(Ch, Q), + delete(Ch, Q), + amqp_connection:close(Conn), + passed. + +ram_duration_test() -> + QName = rabbit_misc:r(<<"/">>, queue, <<"pseudo">>), + Q0 = rabbit_amqqueue:pseudo_queue(QName, self()), + Q = Q0#amqqueue{arguments = [{<<"x-max-priority">>, long, 5}]}, + PQ = rabbit_priority_queue, + BQS1 = PQ:init(Q, new, fun(_, _) -> ok end), + {Duration1, BQS2} = PQ:ram_duration(BQS1), + BQS3 = PQ:set_ram_duration_target(infinity, BQS2), + BQS4 = PQ:set_ram_duration_target(1, BQS3), + {Duration2, BQS5} = PQ:ram_duration(BQS4), + PQ:delete_and_terminate(a_whim, BQS5), + passed. + +mirror_queue_sync_with() -> cluster_ab. +mirror_queue_sync([CfgA, _CfgB]) -> + Ch = pget(channel, CfgA), + Q = <<"test">>, + declare(Ch, Q, 3), + publish(Ch, Q, [1, 2, 3]), + ok = rabbit_test_util:set_ha_policy(CfgA, <<".*">>, <<"all">>), + publish(Ch, Q, [1, 2, 3, 1, 2, 3]), + %% master now has 9, slave 6. + get_partial(Ch, Q, manual_ack, [3, 3, 3, 2, 2, 2]), + %% So some but not all are unacked at the slave + rabbit_test_util:control_action(sync_queue, CfgA, [binary_to_list(Q)], + [{"-p", "/"}]), + wait_for_sync(CfgA, rabbit_misc:r(<<"/">>, queue, Q)), + passed. + +%%---------------------------------------------------------------------------- + +open() -> + {ok, Conn} = amqp_connection:start(#amqp_params_network{}), + {ok, Ch} = amqp_connection:open_channel(Conn), + {Conn, Ch}. + +declare(Ch, Q, Args) when is_list(Args) -> + amqp_channel:call(Ch, #'queue.declare'{queue = Q, + durable = true, + arguments = Args}); +declare(Ch, Q, Max) -> + declare(Ch, Q, arguments(Max)). + +delete(Ch, Q) -> + amqp_channel:call(Ch, #'queue.delete'{queue = Q}). + +publish(Ch, Q, Ps) -> + amqp_channel:call(Ch, #'confirm.select'{}), + [publish1(Ch, Q, P) || P <- Ps], + amqp_channel:wait_for_confirms(Ch). + +publish_many(_Ch, _Q, 0) -> ok; +publish_many( Ch, Q, N) -> publish1(Ch, Q, random:uniform(5)), + publish_many(Ch, Q, N - 1). + +publish1(Ch, Q, P) -> + amqp_channel:cast(Ch, #'basic.publish'{routing_key = Q}, + #amqp_msg{props = props(P), + payload = priority2bin(P)}). + +props(undefined) -> #'P_basic'{delivery_mode = 2}; +props(P) -> #'P_basic'{priority = P, + delivery_mode = 2}. + +consume(Ch, Q, Ack) -> + amqp_channel:subscribe(Ch, #'basic.consume'{queue = Q, + no_ack = Ack =:= no_ack, + consumer_tag = <<"ctag">>}, + self()), + receive + #'basic.consume_ok'{consumer_tag = <<"ctag">>} -> + ok + end. + +cancel(Ch) -> + amqp_channel:call(Ch, #'basic.cancel'{consumer_tag = <<"ctag">>}). + +assert_delivered(Ch, Ack, P) -> + PBin = priority2bin(P), + receive + {#'basic.deliver'{delivery_tag = DTag}, #amqp_msg{payload = PBin2}} -> + ?assertEqual(PBin, PBin2), + maybe_ack(Ch, Ack, DTag) + end. + +get_all(Ch, Q, Ack, Ps) -> + DTags = get_partial(Ch, Q, Ack, Ps), + get_empty(Ch, Q), + DTags. + +get_partial(Ch, Q, Ack, Ps) -> + [get_ok(Ch, Q, Ack, P) || P <- Ps]. + +get_empty(Ch, Q) -> + #'basic.get_empty'{} = amqp_channel:call(Ch, #'basic.get'{queue = Q}). + +get_ok(Ch, Q, Ack, P) -> + PBin = priority2bin(P), + {#'basic.get_ok'{delivery_tag = DTag}, #amqp_msg{payload = PBin2}} = + amqp_channel:call(Ch, #'basic.get'{queue = Q, + no_ack = Ack =:= no_ack}), + ?assertEqual(PBin, PBin2), + maybe_ack(Ch, Ack, DTag). + +maybe_ack(Ch, do_ack, DTag) -> + amqp_channel:cast(Ch, #'basic.ack'{delivery_tag = DTag}), + DTag; +maybe_ack(_Ch, _, DTag) -> + DTag. + +arguments(none) -> []; +arguments(Max) -> [{<<"x-max-priority">>, byte, Max}]. + +priority2bin(undefined) -> <<"undefined">>; +priority2bin(Int) -> list_to_binary(integer_to_list(Int)). + +%%---------------------------------------------------------------------------- + +wait_for_sync(Cfg, Q) -> + case synced(Cfg, Q) of + true -> ok; + false -> timer:sleep(100), + wait_for_sync(Cfg, Q) + end. + +synced(Cfg, Q) -> + Info = rpc:call(pget(node, Cfg), + rabbit_amqqueue, info_all, + [<<"/">>, [name, synchronised_slave_pids]]), + [SSPids] = [Pids || [{name, Q1}, {synchronised_slave_pids, Pids}] <- Info, + Q =:= Q1], + length(SSPids) =:= 1. + +%%----------------------------------------------------------------------------