This is an automated email from the ASF dual-hosted git repository.
jmjoy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking-php.git
The following commit(s) were added to refs/heads/master by this push:
new 83ec158 Add php-amqp plugin (#146)
83ec158 is described below
commit 83ec158150f70ce6be770b127d7ee54990eaf4dc
Author: jmjoy <[email protected]>
AuthorDate: Fri Jun 26 15:51:14 2026 +0800
Add php-amqp plugin (#146)
---
.github/workflows/rust.yml | 4 +-
Cargo.lock | 20 ++--
Cargo.toml | 4 +-
README.md | 2 +-
dist-material/LICENSE | 10 +-
src/execute.rs | 4 +-
src/plugin/mod.rs | 2 +
src/plugin/plugin_amqp.rs | 203 +++++++++++++++++++++++++++++++++++++++
src/plugin/plugin_memcache.rs | 2 +-
src/plugin/plugin_mysqli.rs | 40 ++++----
src/plugin/plugin_psr3.rs | 18 +++-
src/plugin/style.rs | 16 +--
tests/data/expected_context.yaml | 94 +++++++++++++++++-
tests/e2e.rs | 9 ++
tests/php/fpm/amqp.php | 128 ++++++++++++++++++++++++
15 files changed, 499 insertions(+), 57 deletions(-)
diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index d92b69c..c4fb562 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -95,7 +95,7 @@ jobs:
- name: Install dependencies for Linux
if: matrix.os == 'ubuntu-24.04'
- run: sudo apt-get update && sudo apt-get install -y llvm-18-dev
libclang-18-dev protobuf-compiler libsasl2-dev
+ run: sudo apt-get update && sudo apt-get install -y llvm-18-dev
libclang-18-dev protobuf-compiler libsasl2-dev librabbitmq-dev
- name: Install protobuf for Macos
if: matrix.os == 'macos-14'
@@ -110,7 +110,7 @@ jobs:
bcmath, calendar, ctype, dom, exif, gettext, iconv, intl, json,
mbstring,
mysqli, mysqlnd, opcache, pdo, pdo_mysql, phar, posix, readline,
redis,
memcached, swoole-${{ matrix.flag.swoole_version }}, xml,
xmlreader, xmlwriter,
- yaml, zip, mongodb, memcache
+ yaml, zip, mongodb, memcache, amqp
- name: Setup php-fpm for Linux
if: matrix.os == 'ubuntu-24.04'
diff --git a/Cargo.lock b/Cargo.lock
index 5f6ac84..cb6a6f8 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1695,9 +1695,9 @@ dependencies = [
[[package]]
name = "phper"
-version = "0.17.2"
+version = "0.17.5"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "97a3055aa0aec518814199bf9d114eeec267ec89b5008b65c2453ce2cd3ab993"
+checksum = "5a6ded58ef5936b049dcf4b73307889a21b49b9ebd51069f0a554c75275064a7"
dependencies = [
"cfg-if",
"derive_more",
@@ -1712,9 +1712,9 @@ dependencies = [
[[package]]
name = "phper-alloc"
-version = "0.16.1"
+version = "0.16.3"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "0a188ae2a830b7552466093a9acc24e620a6a8678f0bb00384eac8e05cba66c8"
+checksum = "717915904f6ee06640b3232d8d409d7cf3d55b8fc878c83fbd78f73a10466edb"
dependencies = [
"phper-build",
"phper-sys",
@@ -1722,18 +1722,18 @@ dependencies = [
[[package]]
name = "phper-build"
-version = "0.15.4"
+version = "0.15.6"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "3158c29309413a44b2bb80c17d0da09b53df071284f0c4687a232f0787f56a38"
+checksum = "6996befb91ef0538a35dd59f712c9667ad52905750bff74059886e1061a899e2"
dependencies = [
"phper-sys",
]
[[package]]
name = "phper-macros"
-version = "0.15.2"
+version = "0.15.3"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "8dc718c423b154a612117abcc3f0978a8ce9e295d6a80dba6b1b3a7fad3e0ddf"
+checksum = "82de7b26c685b446a2f0e2e718aa36b198cf806b094b7316d04b55b950c0a56a"
dependencies = [
"proc-macro2",
"quote",
@@ -1742,9 +1742,9 @@ dependencies = [
[[package]]
name = "phper-sys"
-version = "0.15.4"
+version = "0.15.6"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "758c37b62c9aa4e4aaf1a04ffdd7631b886fc7086db651fe3d4c0097bc101428"
+checksum = "cd0c78f99806ddfa651f0e1fc7aa86ab616a1819fb8335067f4cc2fb52f7c881"
dependencies = [
"bindgen",
"cc",
diff --git a/Cargo.toml b/Cargo.toml
index b5a6c0a..6c646e7 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -60,7 +60,7 @@ futures-util = "0.3.31"
hostname = "0.4.1"
libc = "0.2.174"
once_cell = "1.21.3"
-phper = "0.17.2"
+phper = "0.17.5"
prost = "0.14.1"
rdkafka = { workspace = true, optional = true }
skywalking = { version = "0.10.0", features = ["management"] }
@@ -81,7 +81,7 @@ fastcgi-client = "0.9.0"
reqwest = { version = "0.12.22", features = ["trust-dns", "json", "stream"] }
[build-dependencies]
-phper-build = "0.15.3"
+phper-build = "0.15.5"
[lints.rust]
unexpected_cfgs = { level = "warn", check-cfg = ['cfg(phper_major_version,
values("8"))'] }
diff --git a/README.md b/README.md
index 60c8415..7fb66bb 100644
--- a/README.md
+++ b/README.md
@@ -31,7 +31,7 @@ SkyWalking PHP Agent requires SkyWalking 8.4+ and PHP 7.2+
* [x] [MySQL Improved](https://www.php.net/manual/en/book.mysqli.php)
* [x] [Memcached](https://www.php.net/manual/en/book.memcached.php)
* [x] [phpredis](https://github.com/phpredis/phpredis)
- * [ ] [php-amqp](https://github.com/php-amqp/php-amqp)
+ * [x] [php-amqp](https://github.com/php-amqp/php-amqp) for Message Queuing
Producer
* [ ] [php-rdkafka](https://github.com/arnaud-lb/php-rdkafka)
* [x] [predis](https://github.com/predis/predis)
* [x] [php-amqplib](https://github.com/php-amqplib/php-amqplib) for Message
Queuing Producer
diff --git a/dist-material/LICENSE b/dist-material/LICENSE
index 33ed232..89cfb69 100644
--- a/dist-material/LICENSE
+++ b/dist-material/LICENSE
@@ -654,11 +654,11 @@ MulanPSL-2.0 licenses
The following components are provided under the MulanPSL-2.0 License. See
project link for details.
The text of each license is also included in licenses/LICENSE-[project].txt.
- https://crates.io/crates/phper/0.17.2 0.17.2 MulanPSL-2.0
- https://crates.io/crates/phper-alloc/0.16.1 0.16.1 MulanPSL-2.0
- https://crates.io/crates/phper-build/0.15.4 0.15.4 MulanPSL-2.0
- https://crates.io/crates/phper-macros/0.15.2 0.15.2 MulanPSL-2.0
- https://crates.io/crates/phper-sys/0.15.4 0.15.4 MulanPSL-2.0
+ https://crates.io/crates/phper/0.17.5 0.17.5 MulanPSL-2.0
+ https://crates.io/crates/phper-alloc/0.16.3 0.16.3 MulanPSL-2.0
+ https://crates.io/crates/phper-build/0.15.6 0.15.6 MulanPSL-2.0
+ https://crates.io/crates/phper-macros/0.15.3 0.15.3 MulanPSL-2.0
+ https://crates.io/crates/phper-sys/0.15.6 0.15.6 MulanPSL-2.0
========================================================================
Unicode-3.0 licenses
diff --git a/src/execute.rs b/src/execute.rs
index 04c7449..9b1cf23 100644
--- a/src/execute.rs
+++ b/src/execute.rs
@@ -321,8 +321,8 @@ fn infer_request_id(execute_data: &mut ExecuteData) ->
Option<i64> {
prev_execute_data_ptr = unsafe {
(*prev_execute_data_ptr).prev_execute_data };
continue;
}
- let request = prev_execute_data.get_mut_parameter(0).as_mut_z_obj()?;
- match request.get_mut_property("fd").as_long() {
+ let request = prev_execute_data.get_parameter(0).as_z_obj()?;
+ match request.get_property("fd").as_long() {
Some(fd) => return Some(fd),
None => {
error!("infer request id failed");
diff --git a/src/plugin/mod.rs b/src/plugin/mod.rs
index 4f7623e..9f97e22 100644
--- a/src/plugin/mod.rs
+++ b/src/plugin/mod.rs
@@ -13,6 +13,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
+mod plugin_amqp;
mod plugin_amqplib;
mod plugin_curl;
mod plugin_memcache;
@@ -48,6 +49,7 @@ static PLUGINS: Lazy<Vec<Box<DynPlugin>>> = Lazy::new(|| {
Box::<plugin_predis::PredisPlugin>::default(),
Box::<plugin_memcached::MemcachedPlugin>::default(),
Box::<plugin_redis::RedisPlugin>::default(),
+ Box::<plugin_amqp::AmqpPlugin>::default(),
Box::<plugin_amqplib::AmqplibPlugin>::default(),
Box::<plugin_mongodb::MongodbPlugin>::default(),
Box::<plugin_memcache::MemcachePlugin>::default(),
diff --git a/src/plugin/plugin_amqp.rs b/src/plugin/plugin_amqp.rs
new file mode 100644
index 0000000..61cbced
--- /dev/null
+++ b/src/plugin/plugin_amqp.rs
@@ -0,0 +1,203 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// 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.
+
+use super::{Plugin, log_exception};
+use crate::{
+ component::COMPONENT_AMQP_PRODUCER_ID,
+ context::{RequestContext, SW_HEADER},
+ execute::{AfterExecuteHook, BeforeExecuteHook, get_this_mut,
validate_num_args},
+ tag::{TAG_MQ_BROKER, TAG_MQ_QUEUE, TAG_MQ_TOPIC},
+};
+use phper::{
+ arrays::ZArray,
+ objects::ZObj,
+ values::{ExecuteData, ZVal},
+};
+use skywalking::{
+ proto::v3::SpanLayer,
+ trace::span::{HandleSpanObject, Span},
+};
+
+#[derive(Default, Clone)]
+pub struct AmqpPlugin;
+
+impl Plugin for AmqpPlugin {
+ fn class_names(&self) -> Option<&'static [&'static str]> {
+ Some(&["AMQPExchange"])
+ }
+
+ fn function_name_prefix(&self) -> Option<&'static str> {
+ None
+ }
+
+ fn hook(
+ &self, class_name: Option<&str>, function_name: &str,
+ ) -> Option<(
+ Box<crate::execute::BeforeExecuteHook>,
+ Box<crate::execute::AfterExecuteHook>,
+ )> {
+ match (class_name, function_name) {
+ (Some(class_name @ "AMQPExchange"), function_name @ "publish") => {
+ Some(self.hook_exchange_publish(class_name, function_name))
+ }
+ _ => None,
+ }
+ }
+}
+
+impl AmqpPlugin {
+ fn hook_exchange_publish(
+ &self, class_name: &str, function_name: &str,
+ ) -> (Box<BeforeExecuteHook>, Box<AfterExecuteHook>) {
+ let class_name = class_name.to_owned();
+ let function_name = function_name.to_owned();
+ (
+ Box::new(move |request_id, execute_data| {
+ validate_num_args(execute_data, 1)?;
+
+ let this = get_this_mut(execute_data)?;
+
+ let peer = Self::get_peer(this)?;
+
+ let exchange = this
+ .call("getName", [])
+ .ok()
+ .and_then(|v| {
+ v.as_z_str()
+ .and_then(|s| s.to_str().ok())
+ .map(ToOwned::to_owned)
+ })
+ .unwrap_or_default();
+
+ let routing_key = if execute_data.num_args() >= 2 {
+ execute_data
+ .get_parameter(1)
+ .as_z_str()
+ .and_then(|s| s.to_str().ok())
+ .map(ToOwned::to_owned)
+ .unwrap_or_default()
+ } else {
+ String::new()
+ };
+
+ let span = Self::create_exit_span(
+ request_id,
+ &class_name,
+ &function_name,
+ &peer,
+ &exchange,
+ &routing_key,
+ )?;
+
+ Self::inject_sw_header(request_id, execute_data, &peer)?;
+
+ Ok(Box::new(span))
+ }),
+ Box::new(move |_, span, _, _| {
+ let mut span = span.downcast::<Span>().unwrap();
+ log_exception(&mut *span);
+ Ok(())
+ }),
+ )
+ }
+
+ fn get_peer(this: &mut ZObj) -> crate::Result<String> {
+ let mut channel = this.call("getChannel", [])?;
+ let channel = channel
+ .expect_mut_z_obj()
+ .map_err(|e| anyhow::anyhow!("channel isn't object: {}", e))?;
+ let mut connection = channel.call("getConnection", [])?;
+ let connection = connection
+ .expect_mut_z_obj()
+ .map_err(|e| anyhow::anyhow!("connection isn't object: {}", e))?;
+ let host = connection.call("getHost", [])?;
+ let host = host
+ .expect_z_str()
+ .map_err(|e| anyhow::anyhow!("host isn't string: {}", e))?
+ .to_str()?;
+ let port = connection.call("getPort", [])?;
+ let port = port.as_long().unwrap_or_default();
+ Ok(format!("{}:{}", host, port))
+ }
+
+ fn create_exit_span(
+ request_id: Option<i64>, class_name: &str, function_name: &str, peer:
&str, exchange: &str,
+ routing_key: &str,
+ ) -> crate::Result<Span> {
+ let mut span = RequestContext::try_with_global_ctx(request_id, |ctx| {
+ Ok(ctx.create_exit_span(&format!("{}->{}", class_name,
function_name), peer))
+ })?;
+
+ let span_object = span.span_object_mut();
+ span_object.set_span_layer(SpanLayer::Mq);
+ span_object.component_id = COMPONENT_AMQP_PRODUCER_ID;
+ span_object.add_tag(TAG_MQ_BROKER, peer);
+ span_object.add_tag(TAG_MQ_TOPIC, exchange);
+ span_object.add_tag(TAG_MQ_QUEUE, routing_key);
+
+ Ok(span)
+ }
+
+ fn inject_sw_header(
+ request_id: Option<i64>, execute_data: &mut ExecuteData, peer: &str,
+ ) -> crate::Result<()> {
+ let sw_header = RequestContext::try_get_sw_header(request_id, peer)?;
+
+ let attributes = Self::ensure_attributes(execute_data)?;
+ let headers = Self::ensure_headers(attributes)?;
+ headers.insert(SW_HEADER, sw_header);
+
+ Ok(())
+ }
+
+ fn ensure_attributes(
+ execute_data: &mut ExecuteData,
+ ) -> crate::Result<&mut phper::arrays::ZArr> {
+ // php-amqp parses publish() as `s|s!l!a/`, so the 4th `headers`
+ // argument is only visible after the missing optional parameters are
+ // materialized and `num_args` reflects that.
+ execute_data.materialize_missing([
+ ZVal::from(()),
+ ZVal::from(()),
+ ZVal::from(ZArray::new()),
+ ])?;
+
+ let attributes = execute_data.get_mut_parameter(3);
+ if attributes.as_z_arr().is_none() {
+ *attributes = ZArray::new().into();
+ }
+
+ Ok(attributes
+ .as_mut_z_arr()
+ .ok_or_else(|| anyhow::anyhow!("attributes isn't array"))?)
+ }
+
+ fn ensure_headers(
+ attributes: &mut phper::arrays::ZArr,
+ ) -> crate::Result<&mut phper::arrays::ZArr> {
+ let has_headers = attributes
+ .get("headers")
+ .and_then(|headers| headers.as_z_arr())
+ .is_some();
+ if !has_headers {
+ attributes.insert("headers", ZArray::new());
+ }
+
+ Ok(attributes
+ .get_mut("headers")
+ .and_then(|headers| headers.as_mut_z_arr())
+ .ok_or_else(|| anyhow::anyhow!("headers isn't array"))?)
+ }
+}
diff --git a/src/plugin/plugin_memcache.rs b/src/plugin/plugin_memcache.rs
index 2b6fb6c..027131d 100644
--- a/src/plugin/plugin_memcache.rs
+++ b/src/plugin/plugin_memcache.rs
@@ -200,7 +200,7 @@ impl MemcachePlugin {
.unwrap();
let key = style
- .get_mut_parameter(execute_data, 0)
+ .get_parameter(execute_data, 0)
.as_z_str()
.and_then(|s| s.to_str().ok())
.map(ToOwned::to_owned)
diff --git a/src/plugin/plugin_mysqli.rs b/src/plugin/plugin_mysqli.rs
index 6e5c261..71b3b5d 100644
--- a/src/plugin/plugin_mysqli.rs
+++ b/src/plugin/plugin_mysqli.rs
@@ -94,7 +94,8 @@ impl MySQLImprovedPlugin {
Box::new(move |request_id, execute_data| {
// Sometimes the connection is failed. Therefore, first
assemble the peer from
// the parameters to prevent assembly failure in the after
hook.
- let peer = get_peer_by_parameters(execute_data, style);
+ let peer =
+ get_peer_by_parameters(execute_data,
class_name.as_deref(), &function_name);
let span = create_mysqli_exit_span(
request_id,
@@ -159,10 +160,8 @@ impl MySQLImprovedPlugin {
style,
)?;
- if execute_data.num_args() >= 1 {
- if let Some(statement) =
execute_data.get_parameter(0).as_z_str() {
- span.add_tag("db.statement", statement.to_str()?);
- }
+ if let Some(statement) =
execute_data.get_parameter(0).as_z_str() {
+ span.add_tag("db.statement", statement.to_str()?);
}
Ok(Box::new(span) as _)
@@ -226,21 +225,28 @@ fn get_peer_by_this(this: &mut ZObj) -> Option<String> {
})
}
-fn get_peer_by_parameters(execute_data: &mut ExecuteData, style: ApiStyle) ->
String {
- let mut peer = "".to_owned();
+fn get_peer_by_parameters(
+ execute_data: &ExecuteData, class_name: Option<&str>, function_name: &str,
+) -> String {
+ let (host_index, port_index) = match (class_name, function_name) {
+ (Some("mysqli"), "__construct" | "real_connect") => (0, 4),
+ (None, "mysqli_connect") => (0, 4),
+ (None, "mysqli_real_connect") => (1, 5),
+ _ => return String::new(),
+ };
- if style.validate_num_args(execute_data, 1).is_ok() {
- peer.push_str(
- style
- .get_mut_parameter(execute_data, 0)
- .as_z_str()
- .and_then(|s| s.to_str().ok())
- .unwrap_or_default(),
- );
- }
+ let mut peer = String::new();
+
+ peer.push_str(
+ execute_data
+ .get_parameter(host_index)
+ .as_z_str()
+ .and_then(|s| s.to_str().ok())
+ .unwrap_or_default(),
+ );
if !peer.is_empty() {
- let port = style.get_mut_parameter(execute_data, 4);
+ let port = execute_data.get_parameter(port_index);
#[allow(clippy::manual_map)]
let port = if let Some(port) = port.as_z_str() {
diff --git a/src/plugin/plugin_psr3.rs b/src/plugin/plugin_psr3.rs
index 9e4a2c7..28088c1 100644
--- a/src/plugin/plugin_psr3.rs
+++ b/src/plugin/plugin_psr3.rs
@@ -16,7 +16,7 @@
use super::Plugin;
use crate::{
context::RequestContext,
- execute::{AfterExecuteHook, BeforeExecuteHook, Noop},
+ execute::{AfterExecuteHook, BeforeExecuteHook, Noop, validate_num_args},
log::PsrLogLevel,
module::PSR_LOGGING_LEVEL,
};
@@ -85,8 +85,14 @@ impl Psr3Plugin {
) -> (Box<BeforeExecuteHook>, Box<AfterExecuteHook>) {
(
Box::new(move |request_id, execute_data| {
+ validate_num_args(execute_data, 1)?;
+
let message =
Self::handle_message(execute_data.get_mut_parameter(0))?;
- let context =
Self::handle_context(execute_data.get_mut_parameter(1))?;
+ let context = if execute_data.num_args() >= 2 {
+ Self::handle_context(execute_data.get_mut_parameter(1))?
+ } else {
+ vec![]
+ };
Self::handle_log(
class_name.clone(),
function_name.clone(),
@@ -107,13 +113,19 @@ impl Psr3Plugin {
) -> (Box<BeforeExecuteHook>, Box<AfterExecuteHook>) {
(
Box::new(move |request_id, execute_data| {
+ validate_num_args(execute_data, 2)?;
+
let log_level =
execute_data.get_parameter(0).expect_z_str()?.to_str()?;
let log_level: PsrLogLevel = log_level.into();
if log_level < *PSR_LOGGING_LEVEL {
return Ok(Box::new(()));
}
let message =
Self::handle_message(execute_data.get_mut_parameter(1))?;
- let context =
Self::handle_context(execute_data.get_mut_parameter(2))?;
+ let context = if execute_data.num_args() >= 3 {
+ Self::handle_context(execute_data.get_mut_parameter(2))?
+ } else {
+ vec![]
+ };
Self::handle_log(
class_name.clone(),
function_name.clone(),
diff --git a/src/plugin/style.rs b/src/plugin/style.rs
index 4d99bf6..4258372 100644
--- a/src/plugin/style.rs
+++ b/src/plugin/style.rs
@@ -13,7 +13,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-use crate::execute::{get_this_mut, validate_num_args};
+use crate::execute::get_this_mut;
use anyhow::Context;
use phper::{
objects::ZObj,
@@ -40,22 +40,12 @@ impl ApiStyle {
}
}
- pub fn get_mut_parameter(self, execute_data: &mut ExecuteData, index:
usize) -> &mut ZVal {
+ pub fn get_parameter(self, execute_data: &ExecuteData, index: usize) ->
&ZVal {
let index = match self {
ApiStyle::OO => index,
ApiStyle::Procedural => index + 1,
};
- execute_data.get_mut_parameter(index)
- }
-
- pub fn validate_num_args(
- self, execute_data: &mut ExecuteData, num: usize,
- ) -> anyhow::Result<()> {
- let num = match self {
- ApiStyle::OO => num,
- ApiStyle::Procedural => num + 1,
- };
- validate_num_args(execute_data, num)
+ execute_data.get_parameter(index)
}
pub fn generate_operation_name(self, class_name: Option<&str>,
function_name: &str) -> String {
diff --git a/tests/data/expected_context.yaml b/tests/data/expected_context.yaml
index 40fbc20..4882270 100644
--- a/tests/data/expected_context.yaml
+++ b/tests/data/expected_context.yaml
@@ -15,7 +15,7 @@
segmentItems:
- serviceName: skywalking-agent-test-1
- segmentSize: 20
+ segmentSize: 21
segments:
- segmentId: "not null"
spans:
@@ -1209,6 +1209,98 @@ segmentItems:
- { key: url, value: "http://127.0.0.1:9011/rabbitmq.php"; }
- { key: http.method, value: GET }
- { key: http.status_code, value: "200" }
+ - segmentId: "not null"
+ spans:
+ - operationName: AMQPExchange->publish
+ parentSpanId: 0
+ spanId: 1
+ spanLayer: MQ
+ startTime: gt 0
+ endTime: gt 0
+ componentId: 144
+ isError: false
+ spanType: Exit
+ peer: 127.0.0.1:5672
+ skipAnalysis: false
+ tags:
+ - { key: mq.broker, value: "127.0.0.1:5672" }
+ - { key: mq.topic, value: fanout_exchange_test }
+ - { key: mq.queue, value: "" }
+ - operationName: AMQPExchange->publish
+ parentSpanId: 0
+ spanId: 2
+ spanLayer: MQ
+ startTime: gt 0
+ endTime: gt 0
+ componentId: 144
+ isError: false
+ spanType: Exit
+ peer: 127.0.0.1:5672
+ skipAnalysis: false
+ tags:
+ - { key: mq.broker, value: "127.0.0.1:5672" }
+ - { key: mq.topic, value: "" }
+ - { key: mq.queue, value: queue_test }
+ - operationName: AMQPExchange->publish
+ parentSpanId: 0
+ spanId: 3
+ spanLayer: MQ
+ startTime: gt 0
+ endTime: gt 0
+ componentId: 144
+ isError: false
+ spanType: Exit
+ peer: 127.0.0.1:5672
+ skipAnalysis: false
+ tags:
+ - { key: mq.broker, value: "127.0.0.1:5672" }
+ - { key: mq.topic, value: exchange_test }
+ - { key: mq.queue, value: routing_test }
+ - operationName: AMQPExchange->publish
+ parentSpanId: 0
+ spanId: 4
+ spanLayer: MQ
+ startTime: gt 0
+ endTime: gt 0
+ componentId: 144
+ isError: false
+ spanType: Exit
+ peer: 127.0.0.1:5672
+ skipAnalysis: false
+ tags:
+ - { key: mq.broker, value: "127.0.0.1:5672" }
+ - { key: mq.topic, value: "" }
+ - { key: mq.queue, value: queue_test }
+ - operationName: AMQPExchange->publish
+ parentSpanId: 0
+ spanId: 5
+ spanLayer: MQ
+ startTime: gt 0
+ endTime: gt 0
+ componentId: 144
+ isError: false
+ spanType: Exit
+ peer: 127.0.0.1:5672
+ skipAnalysis: false
+ tags:
+ - { key: mq.broker, value: "127.0.0.1:5672" }
+ - { key: mq.topic, value: "" }
+ - { key: mq.queue, value: queue_test }
+ - operationName: GET:/amqp.php
+ parentSpanId: -1
+ spanId: 0
+ spanLayer: Http
+ startTime: gt 0
+ endTime: gt 0
+ componentId: 8001
+ isError: false
+ spanType: Entry
+ peer: ""
+ skipAnalysis: false
+ tags:
+ - { key: url, value: "http://127.0.0.1:9011/amqp.php"; }
+ - { key: http.method, value: GET }
+ - { key: http.status_code, value: "200" }
- segmentId: "not null"
spans:
- operationName: "MongoDB\\Driver\\Manager->executeCommand"
diff --git a/tests/e2e.rs b/tests/e2e.rs
index a76ce6c..7b325c6 100644
--- a/tests/e2e.rs
+++ b/tests/e2e.rs
@@ -62,6 +62,7 @@ async fn run_e2e() {
request_fpm_memcached().await;
request_fpm_redis().await;
request_fpm_rabbitmq().await;
+ request_fpm_amqp().await;
request_fpm_mongodb().await;
request_fpm_memcache().await;
request_fpm_monolog().await;
@@ -152,6 +153,14 @@ async fn request_fpm_rabbitmq() {
.await;
}
+async fn request_fpm_amqp() {
+ request_common(
+ HTTP_CLIENT.get(format!("http://{}/amqp.php";, PROXY_SERVER_1_ADDRESS)),
+ "ok",
+ )
+ .await;
+}
+
async fn request_fpm_mongodb() {
request_common(
HTTP_CLIENT.get(format!("http://{}/mongodb.php";,
PROXY_SERVER_1_ADDRESS)),
diff --git a/tests/php/fpm/amqp.php b/tests/php/fpm/amqp.php
new file mode 100644
index 0000000..7c30608
--- /dev/null
+++ b/tests/php/fpm/amqp.php
@@ -0,0 +1,128 @@
+<?php
+
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// 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.
+
+function assertTrue(bool $condition, string $message): void
+{
+ if (!$condition) {
+ throw new RuntimeException($message);
+ }
+}
+
+function assertSameValue($expected, $actual, string $message): void
+{
+ if ($expected !== $actual) {
+ throw new RuntimeException(sprintf('%s. expected=%s actual=%s',
$message, var_export($expected, true), var_export($actual, true)));
+ }
+}
+
+function expectEnvelope(AMQPQueue $queue, string $body, callable $assertions):
void
+{
+ $envelope = null;
+ for ($attempt = 0; $attempt < 10; $attempt++) {
+ $envelope = $queue->get(AMQP_AUTOACK);
+ if ($envelope instanceof AMQPEnvelope) {
+ break;
+ }
+ usleep(100000);
+ }
+
+ assertTrue($envelope instanceof AMQPEnvelope, 'expected message in queue');
+ assertSameValue($body, $envelope->getBody(), 'unexpected message body');
+ $assertions($envelope);
+}
+
+
+$connection = new AMQPConnection(['host' => '127.0.0.1', 'port' => 5672,
'login' => 'guest', 'password' => 'guest']);
+$connection->connect();
+$channel = new AMQPChannel($connection);
+
+$queue = new AMQPQueue($channel);
+$queue->setName('queue_test');
+$queue->setFlags(AMQP_NOPARAM);
+$queue->declareQueue();
+
+$exchange = new AMQPExchange($channel);
+$exchange->setName('exchange_test');
+$exchange->setType(AMQP_EX_TYPE_DIRECT);
+$exchange->declareExchange();
+
+$fanoutQueue = new AMQPQueue($channel);
+$fanoutQueue->setName('fanout_queue_test');
+$fanoutQueue->setFlags(AMQP_NOPARAM);
+$fanoutQueue->declareQueue();
+
+$fanoutExchange = new AMQPExchange($channel);
+$fanoutExchange->setName('fanout_exchange_test');
+$fanoutExchange->setType(AMQP_EX_TYPE_FANOUT);
+$fanoutExchange->declareExchange();
+
+$queue->bind('exchange_test', 'routing_test');
+$fanoutQueue->bind('fanout_exchange_test');
+
+$queue->purge();
+$fanoutQueue->purge();
+
+{
+ $exchange = new AMQPExchange($channel);
+ $exchange->setName('fanout_exchange_test');
+ $exchange->publish('One Arg Message');
+
+ expectEnvelope($fanoutQueue, 'One Arg Message', static function
(AMQPEnvelope $envelope): void {
+ assertTrue($envelope->hasHeader('sw8'), 'missing sw8 header for
one-arg publish');
+ });
+}
+
+{
+ $exchange = new AMQPExchange($channel);
+ $exchange->publish('Two Arg Message', 'queue_test');
+
+ expectEnvelope($queue, 'Two Arg Message', static function (AMQPEnvelope
$envelope): void {
+ assertTrue($envelope->hasHeader('sw8'), 'missing sw8 header for
two-arg publish');
+ });
+}
+
+{
+ $exchange = new AMQPExchange($channel);
+ $exchange->setName('exchange_test');
+ $exchange->publish('Three Arg Message', 'routing_test', AMQP_NOPARAM);
+
+ expectEnvelope($queue, 'Three Arg Message', static function (AMQPEnvelope
$envelope): void {
+ assertTrue($envelope->hasHeader('sw8'), 'missing sw8 header for
three-arg publish');
+ });
+}
+
+{
+ $exchange = new AMQPExchange($channel);
+ $exchange->publish('Four Arg Message', 'queue_test', AMQP_NOPARAM,
['headers' => ['foo' => 'bar']]);
+
+ expectEnvelope($queue, 'Four Arg Message', static function (AMQPEnvelope
$envelope): void {
+ assertTrue($envelope->hasHeader('sw8'), 'missing sw8 header for
four-arg publish');
+ assertTrue($envelope->hasHeader('foo'), 'missing custom header for
four-arg publish');
+ assertSameValue('bar', $envelope->getHeader('foo'), 'custom header
should be preserved');
+ });
+}
+
+{
+ $exchange = new AMQPExchange($channel);
+ $exchange->publish('Default Exchange Message', 'queue_test', AMQP_NOPARAM);
+
+ expectEnvelope($queue, 'Default Exchange Message', static function
(AMQPEnvelope $envelope): void {
+ assertTrue($envelope->hasHeader('sw8'), 'missing sw8 header for
default exchange publish');
+ });
+}
+
+echo "ok";
