(name string)
| 289 | } |
| 290 | |
| 291 | func (fm *functionManagerImpl) ConsumeEvent(name string) (contube.Record, error) { |
| 292 | ctx, cancel := context.WithCancel(context.Background()) |
| 293 | defer cancel() |
| 294 | factory, ok := fm.options.tubeFactoryMap[common.MemoryTubeType] |
| 295 | if !ok { |
| 296 | return nil, errors.New("memory tube factory not found") |
| 297 | } |
| 298 | c, err := factory.NewSourceTube(ctx, (&contube.SourceQueueConfig{ |
| 299 | Topics: []string{name}, SubName: "consume-" + strconv.Itoa(rand.Int())}).ToConfigMap()) |
| 300 | if err != nil { |
| 301 | return nil, err |
| 302 | } |
| 303 | return <-c, nil |
| 304 | } |
| 305 | |
| 306 | // GetStateStore returns the state store used by the function manager |
| 307 | // Return nil if no state store is configured |
nothing calls this directly
no test coverage detected