Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 45 additions & 0 deletions new_samples/branch/branch.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
package main

import (
"fmt"
"time"

"go.uber.org/cadence/workflow"
)

const totalBranches = 3

// BranchWorkflow executes multiple activities in parallel and waits for all.
func BranchWorkflow(ctx workflow.Context) error {
logger := workflow.GetLogger(ctx)
logger.Info("BranchWorkflow started")

ao := workflow.ActivityOptions{
ScheduleToStartTimeout: time.Minute,
StartToCloseTimeout: time.Minute,
}
ctx = workflow.WithActivityOptions(ctx, ao)

var futures []workflow.Future
for i := 1; i <= totalBranches; i++ {
input := fmt.Sprintf("branch %d of %d", i, totalBranches)
future := workflow.ExecuteActivity(ctx, BranchActivity, input)
futures = append(futures, future)
}

for _, f := range futures {
if err := f.Get(ctx, nil); err != nil {
return err
}
}

logger.Info("BranchWorkflow completed")
return nil
}

// BranchActivity processes a single branch.
func BranchActivity(input string) (string, error) {
fmt.Printf("BranchActivity: %s\n", input)
return "Result_" + input, nil
}

23 changes: 23 additions & 0 deletions new_samples/branch/generator/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
<!-- THIS IS A GENERATED FILE -->
<!-- PLEASE DO NOT EDIT -->

# Sample Generator

This folder is NOT part of the actual sample. It exists only for contributors who work on this sample. Please disregard it if you are trying to learn about Cadence.

To create a better learning experience for Cadence users, each sample folder is designed to be self contained. Users can view every part of writing and running workflows, including:

* Cadence client initialization
* Worker with workflow and activity registrations
* Workflow starter
* and the workflow code itself

Some samples may have more or fewer parts depending on what they need to demonstrate.

In most cases, the workflow code (e.g. `workflow.go`) is the part that users care about. The rest is boilerplate needed to run that workflow. For each sample folder, the workflow code should be written by hand. The boilerplate can be generated. Keeping all parts inside one folder gives early learners more value because they can see everything together rather than jumping across directories.

## Contributing

* When creating a new sample, follow the steps mentioned in the README file in the main samples folder.
* To update the sample workflow code, edit the workflow file directly.
* To update the worker, client, or other boilerplate logic, edit the generator file. If your change applies to all samples, update the common generator file inside the `template` folder. Edit the generator file in this folder only when the change should affect this sample alone.
29 changes: 29 additions & 0 deletions new_samples/branch/generator/README_specific.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
## Branch Workflow Sample

This sample demonstrates **parallel activity execution** using Futures.

### Start the Workflow

```bash
cadence --env development \
--domain cadence-samples \
workflow start \
--tl cadence-samples-worker \
--et 60 \
--workflow_type cadence_samples.BranchWorkflow
```

### Key Concept: Parallel Execution with Futures

```go
var futures []workflow.Future
for i := 1; i <= 3; i++ {
future := workflow.ExecuteActivity(ctx, BranchActivity, input)
futures = append(futures, future)
}
// Wait for all
for _, f := range futures {
f.Get(ctx, nil)
}
```

13 changes: 13 additions & 0 deletions new_samples/branch/generator/generate.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
package main

import "github.com/uber-common/cadence-samples/new_samples/template"

func main() {
data := template.TemplateData{
SampleName: "Branch",
Workflows: []string{"BranchWorkflow"},
Activities: []string{"BranchActivity"},
}
template.GenerateAll(data)
}

20 changes: 20 additions & 0 deletions new_samples/branch/main.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
// THIS IS A GENERATED FILE
// PLEASE DO NOT EDIT

package main

import (
"fmt"
"os"
"os/signal"
"syscall"
)

func main() {
StartWorker()

done := make(chan os.Signal, 1)
signal.Notify(done, syscall.SIGINT)
fmt.Println("Cadence worker started, press ctrl+c to terminate...")
<-done
}
101 changes: 101 additions & 0 deletions new_samples/branch/worker.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
// THIS IS A GENERATED FILE
// PLEASE DO NOT EDIT

// Package worker implements a Cadence worker with basic configurations.
package main

import (
"github.com/uber-go/tally"
apiv1 "github.com/uber/cadence-idl/go/proto/api/v1"
"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
"go.uber.org/cadence/activity"
"go.uber.org/cadence/compatibility"
"go.uber.org/cadence/worker"
"go.uber.org/cadence/workflow"
"go.uber.org/yarpc"
"go.uber.org/yarpc/peer"
yarpchostport "go.uber.org/yarpc/peer/hostport"
"go.uber.org/yarpc/transport/grpc"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
)

const (
HostPort = "127.0.0.1:7833"
Domain = "cadence-samples"
// TaskListName identifies set of client workflows, activities, and workers.
// It could be your group or client or application name.
TaskListName = "cadence-samples-worker"
ClientName = "cadence-samples-worker"
CadenceService = "cadence-frontend"
)

// StartWorker creates and starts a basic Cadence worker.
func StartWorker() {
logger, cadenceClient := BuildLogger(), BuildCadenceClient()
workerOptions := worker.Options{
Logger: logger,
MetricsScope: tally.NewTestScope(TaskListName, nil),
}

w := worker.New(
cadenceClient,
Domain,
TaskListName,
workerOptions)
// HelloWorld workflow registration
w.RegisterWorkflowWithOptions(BranchWorkflow, workflow.RegisterOptions{Name: "cadence_samples.BranchWorkflow"})
w.RegisterActivityWithOptions(BranchActivity, activity.RegisterOptions{Name: "cadence_samples.BranchActivity"})

err := w.Start()
if err != nil {
panic("Failed to start worker: " + err.Error())
}
logger.Info("Started Worker.", zap.String("worker", TaskListName))

}

func BuildCadenceClient(dialOptions ...grpc.DialOption) workflowserviceclient.Interface {
grpcTransport := grpc.NewTransport()
// Create a single peer chooser that identifies the host/port and configures
// a gRPC dialer with TLS credentials
myChooser := peer.NewSingle(
yarpchostport.Identify(HostPort),
grpcTransport.NewDialer(dialOptions...),
)
outbound := grpcTransport.NewOutbound(myChooser)

dispatcher := yarpc.NewDispatcher(yarpc.Config{
Name: ClientName,
Outbounds: yarpc.Outbounds{
CadenceService: {Unary: outbound},
},
})
if err := dispatcher.Start(); err != nil {
panic("Failed to start dispatcher: " + err.Error())
}

clientConfig := dispatcher.ClientConfig(CadenceService)

// Create a compatibility adapter that wraps proto-based YARPC clients
// to provide a unified interface for domain, workflow, worker, and visibility APIs
return compatibility.NewThrift2ProtoAdapter(
apiv1.NewDomainAPIYARPCClient(clientConfig),
apiv1.NewWorkflowAPIYARPCClient(clientConfig),
apiv1.NewWorkerAPIYARPCClient(clientConfig),
apiv1.NewVisibilityAPIYARPCClient(clientConfig),
)
}

func BuildLogger() *zap.Logger {
config := zap.NewDevelopmentConfig()
config.Level.SetLevel(zapcore.InfoLevel)

var err error
logger, err := config.Build()
if err != nil {
panic("Failed to setup logger: " + err.Error())
}

return logger
}
93 changes: 93 additions & 0 deletions new_samples/cancelactivity/cancelactivity.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
package main

import (
"context"
"fmt"
"time"

"go.uber.org/cadence"
"go.uber.org/cadence/activity"
"go.uber.org/cadence/workflow"
"go.uber.org/zap"
)

/**
* This is the cancel activity workflow sample.
*/

// ApplicationName is the task list for this sample
const ApplicationName = "cancelGroup"

// CancelWorkflow workflow decider
func CancelWorkflow(ctx workflow.Context) (retError error) {
ao := workflow.ActivityOptions{
ScheduleToStartTimeout: time.Minute,
StartToCloseTimeout: time.Minute * 30,
HeartbeatTimeout: time.Second * 5,
WaitForCancellation: true,
}
ctx = workflow.WithActivityOptions(ctx, ao)
logger := workflow.GetLogger(ctx)
logger.Info("cancel workflow started")

defer func() {
if cadence.IsCanceledError(retError) {
// When workflow is canceled, it has to get a new disconnected context to execute any activities
newCtx, _ := workflow.NewDisconnectedContext(ctx)
err := workflow.ExecuteActivity(newCtx, CleanupActivity).Get(ctx, nil)
if err != nil {
logger.Error("Cleanup activity failed", zap.Error(err))
retError = err
return
}
retError = nil
logger.Info("Workflow completed.")
}
}()

var result string
err := workflow.ExecuteActivity(ctx, LongRunningActivity).Get(ctx, &result)
if err != nil && !cadence.IsCanceledError(err) {
logger.Error("Error from LongRunningActivity", zap.Error(err))
return err
}
logger.Info(fmt.Sprintf("LongRunningActivity returns %v, %v", result, err))

// Execute activity using a canceled ctx,
// activity won't be scheduled and an cancelled error will be returned
err = workflow.ExecuteActivity(ctx, SkippedActivity).Get(ctx, nil)
if err != nil && !cadence.IsCanceledError(err) {
logger.Error("Error from SkippedActivity", zap.Error(err))
}

return err
}

func LongRunningActivity(ctx context.Context) (string, error) {
logger := activity.GetLogger(ctx)
logger.Info("activity started, to cancel workflow, use ./cancelactivity -m cancel -w <WorkflowID> or CLI: 'cadence --do default wf cancel -w <WorkflowID>' to cancel")
for {
select {
case <-time.After(1 * time.Second):
logger.Info("heartbeating...")
activity.RecordHeartbeat(ctx, "")
case <-ctx.Done():
logger.Info("context is cancelled")
// returned canceled error here so that in workflow history we can see ActivityTaskCanceled event
// or if not cancelled, return timeout error
return "I am canceled by Done", ctx.Err()
}
}
}

func CleanupActivity(ctx context.Context) error {
logger := activity.GetLogger(ctx)
logger.Info("CleanupActivity started")
return nil
}

func SkippedActivity(ctx context.Context) error {
logger := activity.GetLogger(ctx)
logger.Info("this activity will be skipped due to cancellation")
return nil
}
23 changes: 23 additions & 0 deletions new_samples/cancelactivity/generator/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
<!-- THIS IS A GENERATED FILE -->
<!-- PLEASE DO NOT EDIT -->

# Sample Generator

This folder is NOT part of the actual sample. It exists only for contributors who work on this sample. Please disregard it if you are trying to learn about Cadence.

To create a better learning experience for Cadence users, each sample folder is designed to be self contained. Users can view every part of writing and running workflows, including:

* Cadence client initialization
* Worker with workflow and activity registrations
* Workflow starter
* and the workflow code itself

Some samples may have more or fewer parts depending on what they need to demonstrate.

In most cases, the workflow code (e.g. `workflow.go`) is the part that users care about. The rest is boilerplate needed to run that workflow. For each sample folder, the workflow code should be written by hand. The boilerplate can be generated. Keeping all parts inside one folder gives early learners more value because they can see everything together rather than jumping across directories.

## Contributing

* When creating a new sample, follow the steps mentioned in the README file in the main samples folder.
* To update the sample workflow code, edit the workflow file directly.
* To update the worker, client, or other boilerplate logic, edit the generator file. If your change applies to all samples, update the common generator file inside the `template` folder. Edit the generator file in this folder only when the change should affect this sample alone.
Loading
Loading