Function TestFMWithGRPCRuntime

fs/runtime/grpc/grpc_func_test.go:31–117

(t *testing.T)

)

func TestFMWithGRPCRuntime(t *testing.T) {
	ctx, closeFSReconcile := context.WithCancel(context.Background())
	fsService := NewFSReconcile(ctx)
	addr := "localhost:17401"
	s, err := StartGRPCServer(fsService, addr)
	if err != nil {
		t.Fatal(err)
		return
	}
	defer s.Stop()
	go StartMockGRPCFunc(t, addr)
	select {
	case <-fsService.WaitForReady():
		t.Logf("ready")
	case <-time.After(1 * time.Second):
		t.Fatal("timeout waiting for fs service ready")
		return
	}

	fm, err := fs.NewFunctionManager(
		fs.WithRuntimeFactory("grpc", fsService),
		fs.WithTubeFactory("memory", contube.NewMemoryQueueFactory(ctx)),
	)
	if err != nil {
		t.Fatal(err)
	}

	inputTopic := "input"
	outputTopic := "output"
	f := &model.Function{
		Name: "test",
		Runtime: model.RuntimeConfig{
			Type: "grpc",
			Config: map[string]interface{}{
				"addr": addr,
			},
		},
		Sources: []model.TubeConfig{
			{
				Type: common.MemoryTubeType,
				Config: (&contube.SourceQueueConfig{
					Topics:  []string{inputTopic},
					SubName: "test",
				}).ToConfigMap(),
			},
		},
		Sink: model.TubeConfig{
			Type: common.MemoryTubeType,
			Config: (&contube.SinkQueueConfig{
				Topic: outputTopic,
			}).ToConfigMap(),
		},
		Replicas: 1,
	}

	err = fm.StartFunction(f)
	if err != nil {
		t.Fatal(err)