func (d *dlqMessageHandlerImpl) Merge(
	ctx context.Context,
	lastMessageID int64,
	pageSize int,
	pageToken []byte,
) ([]byte, error) {

	ackLevel, err := d.replicationQueue.GetDLQAckLevel(ctx)
	if err != nil {
		return nil, err
	}

	messages, token, err := d.replicationQueue.GetMessagesFromDLQ(
		ctx,
		ackLevel,
		lastMessageID,
		pageSize,
		pageToken,
	)
	if err != nil {
		return nil, err
	}

	var ackedMessageID int64
	for _, message := range messages {
		domainTask := message.GetDomainTaskAttributes()
		if domainTask == nil {
			return nil, &types.InternalServiceError{Message: "Encounter non domain replication task in domain replication queue."}
		}

		// TODO:
		if err := d.replicationHandler.Execute(
			domainTask,
		); err != nil {
			return nil, err
		}
		ackedMessageID = message.SourceTaskID
	}

	if err := d.replicationQueue.RangeDeleteMessagesFromDLQ(
		ctx,
		ackLevel,
		ackedMessageID,
	); err != nil {
		d.logger.Error("failed to delete merged tasks on merging domain DLQ message", tag.Error(err))
		return nil, err
	}
	if err := d.replicationQueue.UpdateDLQAckLevel(ctx, ackedMessageID); err != nil {
		d.logger.Error("failed to update ack level on merging domain DLQ message", tag.Error(err))
	}

	return token, nil
}
// MergeMessages merges domain replication DLQ messages
MergeMessages merges domain replication DLQ messages
// MergeMessages merges domain replication DLQ messages
func (d *dlqMessageHandlerImpl) Merge(
	ctx context.Context,
	lastMessageID int64,
	pageSize int,
	pageToken []byte,
) ([]byte, error) {
	ackLevel, err := d.replicationQueue.GetDLQAckLevel(ctx)
	if err != nil {
		return nil, err
	}
	messages, token, err := d.replicationQueue.GetMessagesFromDLQ(
		ctx,
		ackLevel,
		lastMessageID,
		pageSize,
		pageToken,
	)
	if err != nil {
		return nil, err
	}
	var ackedMessageID int64
	for _, message := range messages {
		domainTask := message.GetDomainTaskAttributes()
		if domainTask == nil {
			return nil, &types.InternalServiceError{Message: "Encounter non domain replication task in domain replication queue."}
		}
		if err := d.replicationHandler.Execute(
			domainTask,
		); err != nil {
			return nil, err
		}
		ackedMessageID = message.SourceTaskID
	}
	if err := d.replicationQueue.RangeDeleteMessagesFromDLQ(
		ctx,
		ackLevel,
		ackedMessageID,
	); err != nil {
		d.logger.Error("failed to delete merged tasks on merging domain DLQ message", tag.Error(err))
		return nil, err
	}
	if err := d.replicationQueue.UpdateDLQAckLevel(ctx, ackedMessageID); err != nil {
		d.logger.Error("failed to update ack level on merging domain DLQ message", tag.Error(err))
	}
	return token, nil
}
func ( d * dlqMessageHandlerImpl ) Merge ( ctx context . Context , lastMessageID int64 , pageSize int , pageToken [ ] byte , ) ( [ ] byte , error ) { ackLevel , err := d . replicationQueue . GetDLQAckLevel ( ctx ) if err != nil { return nil , err } messages , token , err := d . replicationQueue . GetMessagesFromDLQ ( ctx , ackLevel , lastMessageID , pageSize , pageToken , ) if err != nil { return nil , err } var ackedMessageID int64 for _ , message := range messages { domainTask := message . GetDomainTaskAttributes ( ) if domainTask == nil { return nil , & types . InternalServiceError { Message : " " } } if err := d . replicationHandler . Execute ( domainTask , ) ; err != nil { return nil , err } ackedMessageID = message . SourceTaskID } if err := d . replicationQueue . RangeDeleteMessagesFromDLQ ( ctx , ackLevel , ackedMessageID , ) ; err != nil { d . logger . Error ( " " , tag . Error ( err ) ) return nil , err } if err := d . replicationQueue . UpdateDLQAckLevel ( ctx , ackedMessageID ) ; err != nil { d . logger . Error ( " " , tag . Error ( err ) ) } return token , nil }
MergeMessages merges domain replication DLQ messages
MergeMessages merges domain replication DLQ messages
// TODO: