Method extractQueueMetadata

common/persistence/sql/queue_v2.go:350–371

(metadataRow *sqlplugin.QueueV2MetadataRow)

}

func (q queueV2) extractQueueMetadata(metadataRow *sqlplugin.QueueV2MetadataRow) (*persistencespb.Queue, error) {
	if metadataRow.MetadataEncoding != enumspb.ENCODING_TYPE_PROTO3.String() {
		return nil, fmt.Errorf(
			"queue with type %v and name %v has invalid encoding: %w",
			metadataRow.QueueType,
			metadataRow.QueueName,
			serialization.NewUnknownEncodingTypeError(metadataRow.MetadataEncoding, enumspb.ENCODING_TYPE_PROTO3),
		)
	}
	qm := &persistencespb.Queue{}
	err := qm.Unmarshal(metadataRow.MetadataPayload)
	if err != nil {
		return nil, serialization.NewDeserializationError(
			enumspb.ENCODING_TYPE_PROTO3,
			fmt.Errorf("unmarshal payload for queue with type %v and name %v failed: %w",
				metadataRow.QueueType,
				metadataRow.QueueName,
				err),
		)
	}
	return qm, nil
}

func (q *queueV2) getMaxMessageID(ctx context.Context, queueType persistence.QueueV2Type, queueName string, tc sqlplugin.TableCRUD) (int64, bool, error) {
	lastMessageID, err := tc.GetLastEnqueuedMessageIDForUpdateV2(ctx, sqlplugin.QueueV2Filter{