607
shardID int32,
608
workflowEvents *WorkflowEvents,
610
>
if len(workflowEvents.Events) == 0 {
611
return nil, nil // allow update workflow without events
612
}
613
615
>
ShardID: shardID,
616
>
BranchToken: workflowEvents.BranchToken,
617
>
Events: workflowEvents.Events,
618
>
PrevTransactionID: workflowEvents.PrevTxnID,
619
>
TransactionID: workflowEvents.TxnID,
620
>
}
621
>
622
>
if workflowEvents.Events[0].EventId == common.FirstEventID {
624
>
request.Info = BuildHistoryGarbageCleanupInfo(workflowEvents.NamespaceID, workflowEvents.WorkflowID, workflowEvents.RunID)
625
>
}
626
628
}
629
630
func (m *executionManagerImpl) SerializeWorkflowMutation( // unexport
631
input *WorkflowMutation,
633
>
634
>
serializedTasks, err := serializeTasks(m.serializer, input.Tasks)
635
>
if err != nil {
636
return nil, err
637
}
638
640
>
NamespaceID: input.ExecutionInfo.GetNamespaceId(),
641
>
WorkflowID: input.ExecutionInfo.GetWorkflowId(),
642
>
RunID: input.ExecutionState.GetRunId(),
643
>
644
>
UpsertActivityInfos: make(map[int64]*commonpb.DataBlob, len(input.UpsertActivityInfos)),
645
>
DeleteActivityInfos: input.DeleteActivityInfos,
646
>
647
>
UpsertTimerInfos: make(map[string]*commonpb.DataBlob, len(input.UpsertTimerInfos)),
648
>
DeleteTimerInfos: input.DeleteTimerInfos,
649
>
650
>
UpsertChildExecutionInfos: make(map[int64]*commonpb.DataBlob, len(input.UpsertChildExecutionInfos)),
651
>
DeleteChildExecutionInfos: input.DeleteChildExecutionInfos,
652
>
653
>
UpsertRequestCancelInfos: make(map[int64]*commonpb.DataBlob, len(input.UpsertRequestCancelInfos)),
654
>
DeleteRequestCancelInfos: input.DeleteRequestCancelInfos,
655
>
656
>
UpsertSignalInfos: make(map[int64]*commonpb.DataBlob, len(input.UpsertSignalInfos)),
657
>
DeleteSignalInfos: input.DeleteSignalInfos,
658
>
659
>
UpsertChasmNodes: make(map[string]InternalChasmNode, len(input.UpsertChasmNodes)),
660
>
DeleteChasmNodes: input.DeleteChasmNodes,
661
>
662
>
UpsertSignalRequestedIDs: input.UpsertSignalRequestedIDs,
663
>
DeleteSignalRequestedIDs: input.DeleteSignalRequestedIDs,
664
>
665
>
NewBufferedEvents: nil,
666
>
ClearBufferedEvents: input.ClearBufferedEvents,
667
>
668
>
ExecutionInfo: input.ExecutionInfo,
669
>
ExecutionState: input.ExecutionState,
670
>
671
>
Tasks: serializedTasks,
672
>
673
>
Condition: input.Condition,
674
>
DBRecordVersion: input.DBRecordVersion,
675
>
NextEventID: input.NextEventID,
676
>
}
677
>
678
>
result.ExecutionInfoBlob, err = m.serializer.WorkflowExecutionInfoToBlob(input.ExecutionInfo)
679
>
if err != nil {
680
return nil, err
681
}
682
>
result.ExecutionStateBlob, err = m.serializer.WorkflowExecutionStateToBlob(input.ExecutionState)
execution_manager.go
683
>
if err != nil {
684
return nil, err
685
}
686
689
>
if err != nil {
690
return nil, err
691
}
693
}
694
697
>
if err != nil {
698
return nil, err
699
}
701
}
702
705
>
if err != nil {
706
return nil, err
707
}
709
}
710
713
>
if err != nil {
714
return nil, err
715
}
717
}
718
721
>
if err != nil {
722
return nil, err
723
}
725
}
726
728
>
if err != nil {
729
return nil, err
730
}
732
>
733
>
if len(input.NewBufferedEvents) > 0 {
734
>
result.NewBufferedEvents, err = m.serializer.SerializeEvents(input.NewBufferedEvents)
execution_manager.go
735
>
if err != nil {
736
return nil, err
737
}
738
}
739
740
>
result.LastWriteVersion, err = getCurrentBranchLastWriteVersion(input.ExecutionInfo.VersionHistories, input.ExecutionInfo.TransitionHistory)
execution_manager.go
741
>
if err != nil {
742
return nil, err
743
}
745
>
if err != nil {
746
return nil, err
747
}
748
750
}
751
752
func (m *executionManagerImpl) SerializeWorkflowSnapshot( // unexport
753
input *WorkflowSnapshot,
755
>
serializedTasks, err := serializeTasks(m.serializer, input.Tasks)
756
>
if err != nil {
757
return nil, err
758
}
759
761
>
NamespaceID: input.ExecutionInfo.GetNamespaceId(),
762
>
WorkflowID: input.ExecutionInfo.GetWorkflowId(),
763
>
RunID: input.ExecutionState.GetRunId(),
764
>
765
>
ActivityInfos: make(map[int64]*commonpb.DataBlob, len(input.ActivityInfos)),
766
>
TimerInfos: make(map[string]*commonpb.DataBlob, len(input.TimerInfos)),
767
>
ChildExecutionInfos: make(map[int64]*commonpb.DataBlob, len(input.ChildExecutionInfos)),
768
>
RequestCancelInfos: make(map[int64]*commonpb.DataBlob, len(input.RequestCancelInfos)),
769
>
SignalInfos: make(map[int64]*commonpb.DataBlob, len(input.SignalInfos)),
770
>
ChasmNodes: make(map[string]InternalChasmNode, len(input.ChasmNodes)),
771
>
772
>
ExecutionInfo: input.ExecutionInfo,
773
>
ExecutionState: input.ExecutionState,
774
>
SignalRequestedIDs: make(map[string]struct{}),
775
>
776
>
Tasks: serializedTasks,
777
>
778
>
Condition: input.Condition,
779
>
DBRecordVersion: input.DBRecordVersion,
780
>
NextEventID: input.NextEventID,
781
>
}
782
>
783
>
result.ExecutionInfoBlob, err = m.serializer.WorkflowExecutionInfoToBlob(input.ExecutionInfo)
784
>
if err != nil {
785
return nil, err
786
}
787
>
result.ExecutionStateBlob, err = m.serializer.WorkflowExecutionStateToBlob(input.ExecutionState)
execution_manager.go
788
>
if err != nil {
789
return nil, err
790
}
791
>
result.LastWriteVersion, err = getCurrentBranchLastWriteVersion(input.ExecutionInfo.VersionHistories, input.ExecutionInfo.TransitionHistory)
execution_manager.go
792
>
if err != nil {
793
return nil, err
794
}
795
798
>
if err != nil {
799
return nil, err
800
}
802
}
805
>
if err != nil {
806
return nil, err
807
}
809
}
812
>
if err != nil {
813
return nil, err
814
}
816
}
819
>
if err != nil {
820
return nil, err
821
}
823
}
826
>
if err != nil {
827
return nil, err
828
}
830
}
833
>
}
835
>
if err != nil {
836
return nil, err
837
}
839
>
840
>
result.Checksum, err = m.serializer.ChecksumToBlob(input.Checksum)
841
>
if err != nil {
842
return nil, err
843
}
844
846
}
847