Method initializeQueueMetadata

common/persistence/sql/queue.go:315–345

(
	ctx context.Context,
	blob *commonpb.DataBlob,
)

}

func (q *sqlQueue) initializeQueueMetadata(
	ctx context.Context,
	blob *commonpb.DataBlob,
) error {
	_, err := q.DB.SelectFromQueueMetadata(ctx, sqlplugin.QueueMetadataFilter{
		QueueType: q.queueType,
	})
	switch err {
	case nil:
		return nil
	case sql.ErrNoRows:
		result, err := q.DB.InsertIntoQueueMetadata(ctx, &sqlplugin.QueueMetadataRow{
			QueueType:    q.queueType,
			Data:         blob.Data,
			DataEncoding: blob.EncodingType.String(),
		})
		if err != nil {
			return serviceerror.NewUnavailablef("initializeQueueMetadata operation failed. Error %v", err)
		}
		rowsAffected, err := result.RowsAffected()
		if err != nil {
			return fmt.Errorf("rowsAffected returned error when initializing queue metadata  %v: %v", q.queueType, err)
		}
		if rowsAffected != 1 {
			return fmt.Errorf("rowsAffected returned %v queue metadata instead of one", rowsAffected)
		}
		return nil
	default:
		return err
	}
}

func (q *sqlQueue) initializeDLQMetadata(
	ctx context.Context,