( ctx context.Context, tx sqlplugin.Tx, shardID int32, replicationTasks []p.InternalHistoryTask, )
| 864 | } |
| 865 | |
| 866 | func createReplicationTasks( |
| 867 | ctx context.Context, |
| 868 | tx sqlplugin.Tx, |
| 869 | shardID int32, |
| 870 | replicationTasks []p.InternalHistoryTask, |
| 871 | ) error { |
| 872 | |
| 873 | if len(replicationTasks) == 0 { |
| 874 | return nil |
| 875 | } |
| 876 | |
| 877 | replicationTasksRows := make([]sqlplugin.ReplicationTasksRow, 0, len(replicationTasks)) |
| 878 | for _, task := range replicationTasks { |
| 879 | replicationTasksRows = append(replicationTasksRows, sqlplugin.ReplicationTasksRow{ |
| 880 | ShardID: shardID, |
| 881 | TaskID: task.Key.TaskID, |
| 882 | Data: task.Blob.Data, |
| 883 | DataEncoding: task.Blob.EncodingType.String(), |
| 884 | }) |
| 885 | } |
| 886 | |
| 887 | result, err := tx.InsertIntoReplicationTasks(ctx, replicationTasksRows) |
| 888 | if err != nil { |
| 889 | return serviceerror.NewUnavailablef("createReplicationTasks failed. Error: %v", err) |
| 890 | } |
| 891 | |
| 892 | rowsAffected, err := result.RowsAffected() |
| 893 | if err != nil { |
| 894 | return serviceerror.NewUnavailablef("createReplicationTasks failed. Could not verify number of rows inserted. Error: %v", err) |
| 895 | } |
| 896 | |
| 897 | if int(rowsAffected) != len(replicationTasks) { |
| 898 | return serviceerror.NewUnavailablef("createReplicationTasks failed. Inserted %v instead of %v rows into transfer_tasks. Error: %v", rowsAffected, len(replicationTasks), err) |
| 899 | } |
| 900 | return nil |
| 901 | } |
| 902 | |
| 903 | func createVisibilityTasks( |
| 904 | ctx context.Context, |
no test coverage detected