| 141 | } |
| 142 | |
| 143 | func (s *ActorSuite) TestPipeline() { |
| 144 | // 创建 3 阶段流水线 |
| 145 | stage3 := &PipelineStageActor{name: "stage3", stage: 3, nextStage: nil} |
| 146 | pid3 := s.system.Spawn(stage3, "stage3") |
| 147 | |
| 148 | stage2 := &PipelineStageActor{name: "stage2", stage: 2, nextStage: pid3} |
| 149 | pid2 := s.system.Spawn(stage2, "stage2") |
| 150 | |
| 151 | stage1 := &PipelineStageActor{name: "stage1", stage: 1, nextStage: pid2} |
| 152 | pid1 := s.system.Spawn(stage1, "stage1") |
| 153 | |
| 154 | // 发送数据 |
| 155 | resultCh := make(chan string, 1) |
| 156 | pid1.Tell(&ProcessMsg{ |
| 157 | Data: "Input", |
| 158 | Stage: 1, |
| 159 | Result: resultCh, |
| 160 | }) |
| 161 | |
| 162 | // 等待结果 |
| 163 | select { |
| 164 | case result := <-resultCh: |
| 165 | expected := "Input -> Stage1 -> Stage2 -> Stage3" |
| 166 | s.Equal(expected, result, "流水线结果应正确") |
| 167 | case <-time.After(5 * time.Second): |
| 168 | s.T().Fatal("流水线处理超时") |
| 169 | } |
| 170 | } |
| 171 | |
| 172 | func (s *ActorSuite) TestBroadcast() { |
| 173 | // 创建订阅者 |