问题标签 [automatonymous]

我试图了解服务之间消息传递中使用的不同方法。

假设我有一个场景，我需要第一个服务来通知另一个用户已经要求创建产品，第二个服务应该收到这个消息，创建一个产品，然后响应告诉第一个服务一个产品已经创建的。

我认为命令以及请求/响应适合这种情况，因为第一个服务将需要处理另一个特定服务并等待反馈。

我的理解是：

事件与命令：

事件：

• 提供服务之间的松散耦合。
• 向所有队列执行发布，对此类消息感兴趣的服务将选择它。

命令：

• 执行发送到特定队列，因此只有使用该队列接收的服务才会使用它。

请求/响应与发布/订阅：

请求/响应：

在 Request/Respond 中，第一个服务请求另一个服务执行操作并等待后者返回响应。

发布/订阅：

第一个服务只是发布一条消息并继续处理，而无需等待反馈或响应。

现在我开始使用 RabbitMQ 和 Masstransit saga (Masstransit.Automatonymous) 来设计消息系统，它似乎遵循带有发布/订阅方法的事件。

我的问题是：

我可以使用带有发布的命令或带有请求/响应的事件吗？

我的理解正确吗？sagas 可以与请求/响应一起使用吗？

我是服务总线的初学者，并试图理解使用 Masstransit 和 Automatonymous 的 Saga 和 Statemachines 的概念。

我的情况是，一个计算一个县的食物消耗的传奇。

传奇将针对一个地理状态触发。第一个任务是使用 API 调用获取所有县的列表。

然后，为每个县发布/发送食品消费事件。一旦食物消耗计算器完成，它将引发一个事件。

原始 saga 将等待所有这些完成事件，当它收到所有这些事件时，将整理响应并创建最终统计报告。

我能够找到使用此线程 处理多个事件的状态转换来整理响应的能力

但是，我无法找到调用地理状态查找 api 并从状态机中触发单个事件的方法。

我终于开始利用 MassTransit 中出色的 Automatonymous 组件，并且我想通过我的新状态机进行 TDD。

在阅读了此处的 MT 文档 ( http://masstransit-project.com/MassTransit/advanced/sagas/automatonymous.html ) 并花了一些时间谷歌搜索后，我在 MT/Automatonymous Git 存储库中找到了单元测试，看起来像要走的路：

• https://github.com/MassTransit/Automatonymous/blob/master/src/Automatonymous.Tests/Condition_Specs.cs#L21

特别是，_machine.RaiseEvent(....)方法似乎正是我需要能够使用测试提供的刺激来驱动状态机。

接下来，我去寻找那个重载......原来它是一个似乎生活在这里的扩展方法：

https://github.com/MassTransit/Automatonymous/blob/15e9181ed05e37ff71e85b514aafc351b1edf27f/src/Automatonymous/RaiseEventExtensions.tt

这就是我卡住的地方。我不确定使用这些扩展方法进行测试的正确方法是什么？*.tt 是一种文本生成模板格式...这些是作为 nuget 包的一部分安装的吗？如何访问扩展程序？和/或更广泛地说，为继承自的状态机编写单元测试的推荐方法是什么MassTransitStateMachine<T>？

此处的后续问题：如何通过 InMemoryTestHarness 成功驱动 MassTransitStateMachine？

跟进：如何编写 MassTransitStateMachine 单元测试？

这是一个简单的状态机的简单测试类（使用 MS 测试），称为ProcedureStateMachine（注意：这对我们来说不是一个真正的生产状态机......只是我曾经玩过MassTransitStateMachine一段时间的一个实验......看起来一个方便的自包含的地方来尝试进行单元测试）：

这是状态机类本身：

调试此测试时，它的集合中_harness有BeginProcessing消息Sent，但集合中没有任何内容_saga.Created。似乎我缺少一些管道以使线束在发送消息时实际驱动状态机？

====

从测试中删除.Wait()调用SetUp()并将TearDown()测试更新为以下内容不会改变行为：

我正在尝试使用带有 RabbitMQ 的 Automatonymous 为状态机实现一个简单的示例/演示。不幸的是，我找不到一个可以重建/学习的东西（我找到了ShoppingWeb，但在我看来它绝非简单）。在我看来，文档也缺乏信息。

这是我想到的状态机示例（对不起，它很丑）： 请注意，这个示例完全是虚构的，是否有意义并不重要。这个项目的目的是让 Automatonymous 变得“温暖”。

我想做/拥有的是：

• 四个应用程序运行：
  1. 状态机本身
  2. “请求者”发送请求被解释
  3. “验证器”或“解析器”检查提供的请求是否有效
  4. 解释给定请求的“解释器”
• 这方面的一个例子可能是：
  • 请求者发送“x=5”
  • 验证器检查是否包含“=”
  • 口译员说“5”

我的状态机实现如下所示：

然后我有这样的服务：

我的问题是：

1. 如何发送“初始”请求，以便状态机转换到我的初始状态
2. 我如何在消费者中“做出反应”以检查已发送的数据，然后像 1 中一样发送新数据？

我正在尝试为 Automatonymous 状态机编写测试，但我在正确处理它时遇到了一些麻烦，而且我发现的文档很少。

这是我目前进行的一项测试：

鉴于 _Saga.Match 调用找到了匹配项，我希望所有消息都已被处理，并且我应该能够获取我的状态机实例和发布的事件并检查它们的值 - 但事实并非如此。当我在夹具中运行测试时，有时我得到的实例已经消耗并发布了预期的消息；有时它还不完全在那里。

我尝试使用以下方法获取我的实例：

或通过以下方式获取已发布的事件：

但是对 Match 的调用是否成功并不重要，状态机实例（和已发布的事件）并不总是存在。

我假设来自 Automatonymous、MassTransit 或测试工具的异步进程导致了不一致。有什么帮助吗？

使用 MassTransit、MassTransit.Automatonymous 和 MassTransit.TestFramework 5.1.2.1528、Automatonymous 4.1.1.102 进行测试，

编辑：

进一步审查，我发现当我遇到问题时，调用Match( ... ) 没有成功 - 它超时。（我一直错误地假设超时会引发异常。）

我试图通过遵循与在 NServiceBus 中实现 Sagas 的方式类似的模式，在不使用自动状态机（我开始使用但发现很难正确进行单元测试）的情况下手动启动 Saga。

但是，我遇到了一个问题，即在初始消息创建实例之前，消息正在命中 saga，这会导致 MassTransit 默默地吞下消息，而不会引发异常或将消息移动到错误队列中。

在试图找出解决这个问题的过程中，很多人建议使用OnMissingInstance错误消息，然后依靠重试框架来有效地延迟它，直到 saga 被正确初始化。 见这里。

我想知道是否有一种方法可以在不使用 Automatonymous 框架的情况下做到这一点，很可能是通过利用一些中间件来抢先检查以确保在消息尝试处理之前存在 Saga 消息（抛出异常并稍后重试）它？如果不是，这听起来像是有用/可能的吗？

更多信息：

- 使用 Azure 服务总线
-MongoDB 实现 Saga 持久性。

一些代码简化了我想要的代码片段：

我觉得我已经接近完成这项工作，但似乎无法完成...

我有一个带有传奇/状态机的 .NET Core ASP 应用程序，它在大多数情况下似乎运行良好。它：

1. 收到请求
2. 发布由传送单选取的事件
3. 当传送单完成时，它会发布一个事件
4. 事件拾起saga
5. 然后 saga 向请求/响应消费者发送一个请求
6. 回应回到传奇
7. 但是然后我使用 RespondAsync 尝试将响应发送回原始调用控制器，但是没有任何东西可以返回

我的控制器看起来像：

我的传奇故事的精简版如下所示：

最后在我的启动代码中，我将请求/响应配置为：

我猜 RespondAsync 调用没有使用正确/原始的 requestId，但我不知道如何检查或更改它。谁能帮忙？

在启用 UseInMemoryOutbox 的公共交通传奇中，是否有可能丢失事件？据我了解UseInMemoryOutbox，它会在持久化saga后推迟发布事件，那么如果saga持久化然后服务器重启会发生什么？

如果是真的，有什么解决方法可以避免这种情况？

此外，如果我不使用 UseInMemoryOutbox，我将面临 saga 不会为传入事件做任何事情的情况，因为它的状态尚未持久化。

我正在尝试将 Redis 持久性添加到我的传奇中，它正在管理对路由单的调用（以及根据路由单的结果向其他消费者发送的附加消息），希望它能解决我不断收到的另一个超时问题.

但是，我收到一条错误消息，该消息进入 RabbitMQ 中的 saga_error 队列。

消息中显示的错误是：

来自程序集“AutomatonymousGreenPipes.DynamicInternalc83411641fad46798326d78fe60522c9, Version=0.0.0.0, Culture=neutral, PublicKeyToken=null”的“GreenPipes.DynamicInternal.Automatonymous.State”类型中的方法“Accept”没有实现

我的相关配置代码是：

上述代码中的 LinkId 始终是唯一的 Guid。

当 saga 正在读回从我的路由单发送的事件（无论是成功还是失败事件）时，问题似乎就会发生。

一个不工作的示例事件接口是：

如果我切换回 InMemorySagaRepository 一切正常（本地）。我已经尝试了很多不同的组合，现在已经碰壁了。

我已将所有软件包更新到最新版本。我也一直在检查我的 redis 数据库，并且可以看到状态机实例每次都正确进入。

我还看到 Google 群组中的某个人遇到了同样的问题，但他们的帖子没有回复。