<commit_before>﻿// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License.
// You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using Kafka.Batching;
using Kafka.Cluster;
using Kafka.Common;
using Kafka.Protocol;
using Kafka.Public;
using ICluster = Kafka.Cluster.ICluster;

namespace Kafka.Routing
{
    /// <summary>
    /// Expose messages consumed from topics.
    /// </summary>
    interface IConsumeRouter
    {
        /// <summary>
        /// Provides a new routing table.
        /// </summary>
        /// <param name="table">The new table.</param>
        void ChangeRoutingTable(RoutingTable table);

        /// <summary>
        /// Consume a given partition of given topic, starting from the given offset.
        /// </summary>
        /// <param name="topic">Topic required to be consumed.</param>
        /// <param name="partition">The partition to consume. The special value -1 means consume
        /// all partitions.</param>
        /// <param name="offset">Offset to start consuming from. There are two special values:
        /// -1 means start from last available offset, -2 means start from first available offset</param>
        void StartConsume(string topic, int partition, long offset);

        /// <summary>
        /// Start consuming from a consumer group subscription.
        /// </summary>
        /// <param name="group"></param>
        /// <param name="topics"></param>
        void StartConsumeSubscription(IConsumerGroup group, IEnumerable<string> topics);

        /// <summary>
        /// Stop consuming a given topic / partition once the given offset is reached.
        /// </summary>
        /// <param name="topic">Topic to stop consuming from.</param>
        /// <param name="partition">Topic to stop consuming from. -1 means all.</param>
        /// <param name="offset">Stop consuming once this offset is reached. Offsets.Now (aka -42) means now.</param>
        void StopConsume(string topic, int partition, long offset);

        /// <summary>
        /// Commit offsets if linked to a consumer group.
        /// This is a fire and forget commit: the consumer will commit offsets at the next
        /// possible occasion. In particular if this is fired from inside a MessageReceived handler, commit
        /// will occur just after returning from the handler.
        /// Offsets commited are all the the offsets of the messages that have been sent through MessageReceived.
        /// </summary>
        void RequireCommit();

        /// <summary>
        /// Commit the given offset for the given offset/partition. This is a fined grained
        /// asynchronous commit. When this method async returns, given offsets will have
        /// effectively been committed.
        /// </summary>
        /// <param name="topic"></param>
        /// <param name="partition"></param>
        /// <param name="offset"></param>
        /// <returns></returns>
        Task CommitAsync(string topic, int partition, long offset);

        /// <summary>
        /// Acknowledge a response from a fetch request.
        /// </summary>
        /// <param name="fetchResponse">a fetch response from a Kafka broker.</param>
        void Acknowledge(CommonAcknowledgement<FetchResponse> fetchResponse);

        /// <summary>
        /// Acknowledge a response from an offset request.
        /// </summary>
        /// <param name="offsetResponse">an offset response from a Kafka broker.</param>
        void Acknowledge(CommonAcknowledgement<CommonResponse<OffsetPartitionResponse>> offsetResponse);

        /// <summary>
        /// Stop the consumer router.
        /// </summary>
        /// <returns></returns>
        Task Stop();

        /// <summary>
        /// Emit a received kafka message.
        /// </summary>
        event Action<RawKafkaRecord> MessageReceived;

        /// <summary>
        /// Signaled when a fetch response has been throttled
        /// </summary>
        event Action<int> Throttled;
    }

    // Magic values for Offset APIs. Some from Kafka protocol, some special to us
    static class Offsets
    {
        public const long Some = 0; // From Kafka protocol
        public const long Latest = -1; // From Kafka protocol
        public const long Earliest = -2; // From Kafka protocol
        public const long Now = -42;
        public const long None = -43;
        public const long Never = -44;
    }

    /// <summary>
    /// Representation of a Kafka fetch unit.
    /// A Kafka FetchRequest will typically contain multiple fetch units.
    /// </summary>
    struct FetchMessage
    {
        public string Topic;  // Required topic
        public long Offset;   // Fetch from this offset
        public int Partition; // Fetch from this partition
        public int MaxBytes;  // Maximum number of bytes to return from this partition
    }

    /// <summary>
    /// Representation of a Kafka Offset unit.
    /// A Kafka OffsetRequest will typically contain multiple offset units.
    /// This request type is very confusing. Basically you use it to get starting
    /// offsets for segments in partition, but segments are an implementation detail
    /// of the Kafka broker. An external client will probably only use the special
    /// values for Time.
    /// </summary>
    struct OffsetMessage
    {
        // Topic to get offsets from.
        public string Topic;

        // From the Kafka Protocol: "Used to ask for all messages before a certain time (ms).
        // There are two special values. Specify -1 to receive the latest offset (i.e. the
        // offset of the next coming message) and -2 to receive the earliest available offset.
        // Note that because offsets are pulled in descending order, asking for the earliest
        // offset will always return you a single element."
        // We will probably only use the special values.
        public long Time;

        // Partition to get offsets from.
        public int Partition;

        // The maximum number of offsets to return. We will probably use 1.
        public int MaxNumberOfOffsets;
    }

    struct CommonAcknowledgement<TResponse> where TResponse : IMemoryStreamSerializable, new()
    {
        public TResponse Response;
        public DateTime ReceivedDate;
    }

    /// <summary>
    /// Consumer implementation. For now partitions handled by the consumer must be specified,
    /// the offset fetch/commit API is not supported.
    /// </summary>
    class ConsumeRouter : IConsumeRouter
    {
        #region Batching

        class ByNodeFetchBatchingStrategy : IBatchStrategy<FetchMessage>
        {
            public bool Send(INode node, FetchMessage message)
            {
                return node.Fetch(message);
            }

            public void Dispose()
            {
            }
        }

        class GlobalFetchBatchingStrategy : IBatchStrategy<FetchMessage>
        {
            private readonly AccumulatorByNodeByTopic<FetchMessage> _accumulator;
            private Action<INode, IBatchByTopic<FetchMessage>> _batch;

            public GlobalFetchBatchingStrategy(Configuration configuration, Action<INode, IBatchByTopic<FetchMessage>> batch)
            {
                _accumulator = new AccumulatorByNodeByTopic<FetchMessage>(
                    m => m.Topic,
                    configuration.ConsumeBatchSize,
                    configuration.ConsumeBufferingTime);
                _accumulator.NewBatch += batch;
            }

            public bool Send(INode node, FetchMessage message)
            {
                return _accumulator.Add(Tuple.Create(node, message));
            }

            public void Dispose()
            {
                _accumulator.Dispose();
                _accumulator.NewBatch -= _batch;
                _batch = null;
            }
        }

        class ByNodeOffsetBatchingStrategy : IBatchStrategy<OffsetMessage>
        {
            public bool Send(INode node, OffsetMessage message)
            {
                return node.Offset(message);
            }

            public void Dispose()
            {
            }
        }

        class GlobalOffsetBatchingStrategy : IBatchStrategy<OffsetMessage>
        {
            private readonly AccumulatorByNodeByTopic<OffsetMessage> _accumulator;
            private Action<INode, IBatchByTopic<OffsetMessage>> _batch;

            public GlobalOffsetBatchingStrategy(Configuration configuration, Action<INode, IBatchByTopic<OffsetMessage>> batch)
            {
                _accumulator = new AccumulatorByNodeByTopic<OffsetMessage>(
                    m => m.Topic,
                    configuration.ConsumeBatchSize,
                    configuration.ConsumeBufferingTime);
                _accumulator.NewBatch += batch;
            }

            public bool Send(INode node, OffsetMessage message)
            {
                return _accumulator.Add(Tuple.Create(node, message));
            }

            public void Dispose()
            {
                _accumulator.Dispose();
                _accumulator.NewBatch -= _batch;
                _batch = null;
            }
        }

        #endregion

        #region Messages

        private enum ConsumerMessageType
        {
            FetchResponse,
            OffsetResponse,
            CheckPostponed,
            CheckPostponedAfterRoutingTableChange,
            Start,
            Stop,
            StartSubscription,
            Commit,
            Heartbeat,
        }

        private struct ConsumerMessage
        {
            public ConsumerMessageType MessageType;
            public ConsumerMessageValue MessageValue;
        }

        class TopicOnOff
        {
            public string Topic;
            public long Offset;
            public int Partition;
        }

        class CommitMsg
        {
            public TaskCompletionSource<bool> CommitPromise;
            public string Topic;
            public long Offset;
            public int Partition;
        }

        [StructLayout(LayoutKind.Explicit)]
        struct ConsumerMessageValue
        {
            [FieldOffset(0)]
            public CommonAcknowledgement<FetchResponse> FetchResponse;

            [FieldOffset(8)]
            public CommonAcknowledgement<CommonResponse<OffsetPartitionResponse>> OffsetResponse;

            [FieldOffset(8)]
            public IEnumerable<string> Subscription;

            [FieldOffset(8)]
            public TopicOnOff Topic;

            [FieldOffset(8)]
            public CommitMsg CommitMsg;
        }
<commit_msg>Fix alignments in 32 bits.<commit_after>﻿// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License.
// You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using Kafka.Batching;
using Kafka.Cluster;
using Kafka.Common;
using Kafka.Protocol;
using Kafka.Public;
using ICluster = Kafka.Cluster.ICluster;

namespace Kafka.Routing
{
    /// <summary>
    /// Expose messages consumed from topics.
    /// </summary>
    interface IConsumeRouter
    {
        /// <summary>
        /// Provides a new routing table.
        /// </summary>
        /// <param name="table">The new table.</param>
        void ChangeRoutingTable(RoutingTable table);

        /// <summary>
        /// Consume a given partition of given topic, starting from the given offset.
        /// </summary>
        /// <param name="topic">Topic required to be consumed.</param>
        /// <param name="partition">The partition to consume. The special value -1 means consume
        /// all partitions.</param>
        /// <param name="offset">Offset to start consuming from. There are two special values:
        /// -1 means start from last available offset, -2 means start from first available offset</param>
        void StartConsume(string topic, int partition, long offset);

        /// <summary>
        /// Start consuming from a consumer group subscription.
        /// </summary>
        /// <param name="group"></param>
        /// <param name="topics"></param>
        void StartConsumeSubscription(IConsumerGroup group, IEnumerable<string> topics);

        /// <summary>
        /// Stop consuming a given topic / partition once the given offset is reached.
        /// </summary>
        /// <param name="topic">Topic to stop consuming from.</param>
        /// <param name="partition">Topic to stop consuming from. -1 means all.</param>
        /// <param name="offset">Stop consuming once this offset is reached. Offsets.Now (aka -42) means now.</param>
        void StopConsume(string topic, int partition, long offset);

        /// <summary>
        /// Commit offsets if linked to a consumer group.
        /// This is a fire and forget commit: the consumer will commit offsets at the next
        /// possible occasion. In particular if this is fired from inside a MessageReceived handler, commit
        /// will occur just after returning from the handler.
        /// Offsets commited are all the the offsets of the messages that have been sent through MessageReceived.
        /// </summary>
        void RequireCommit();

        /// <summary>
        /// Commit the given offset for the given offset/partition. This is a fined grained
        /// asynchronous commit. When this method async returns, given offsets will have
        /// effectively been committed.
        /// </summary>
        /// <param name="topic"></param>
        /// <param name="partition"></param>
        /// <param name="offset"></param>
        /// <returns></returns>
        Task CommitAsync(string topic, int partition, long offset);

        /// <summary>
        /// Acknowledge a response from a fetch request.
        /// </summary>
        /// <param name="fetchResponse">a fetch response from a Kafka broker.</param>
        void Acknowledge(CommonAcknowledgement<FetchResponse> fetchResponse);

        /// <summary>
        /// Acknowledge a response from an offset request.
        /// </summary>
        /// <param name="offsetResponse">an offset response from a Kafka broker.</param>
        void Acknowledge(CommonAcknowledgement<CommonResponse<OffsetPartitionResponse>> offsetResponse);

        /// <summary>
        /// Stop the consumer router.
        /// </summary>
        /// <returns></returns>
        Task Stop();

        /// <summary>
        /// Emit a received kafka message.
        /// </summary>
        event Action<RawKafkaRecord> MessageReceived;

        /// <summary>
        /// Signaled when a fetch response has been throttled
        /// </summary>
        event Action<int> Throttled;
    }

    // Magic values for Offset APIs. Some from Kafka protocol, some special to us
    static class Offsets
    {
        public const long Some = 0; // From Kafka protocol
        public const long Latest = -1; // From Kafka protocol
        public const long Earliest = -2; // From Kafka protocol
        public const long Now = -42;
        public const long None = -43;
        public const long Never = -44;
    }

    /// <summary>
    /// Representation of a Kafka fetch unit.
    /// A Kafka FetchRequest will typically contain multiple fetch units.
    /// </summary>
    struct FetchMessage
    {
        public string Topic;  // Required topic
        public long Offset;   // Fetch from this offset
        public int Partition; // Fetch from this partition
        public int MaxBytes;  // Maximum number of bytes to return from this partition
    }

    /// <summary>
    /// Representation of a Kafka Offset unit.
    /// A Kafka OffsetRequest will typically contain multiple offset units.
    /// This request type is very confusing. Basically you use it to get starting
    /// offsets for segments in partition, but segments are an implementation detail
    /// of the Kafka broker. An external client will probably only use the special
    /// values for Time.
    /// </summary>
    struct OffsetMessage
    {
        // Topic to get offsets from.
        public string Topic;

        // From the Kafka Protocol: "Used to ask for all messages before a certain time (ms).
        // There are two special values. Specify -1 to receive the latest offset (i.e. the
        // offset of the next coming message) and -2 to receive the earliest available offset.
        // Note that because offsets are pulled in descending order, asking for the earliest
        // offset will always return you a single element."
        // We will probably only use the special values.
        public long Time;

        // Partition to get offsets from.
        public int Partition;

        // The maximum number of offsets to return. We will probably use 1.
        public int MaxNumberOfOffsets;
    }

    class CommonAcknowledgement<TResponse> where TResponse : IMemoryStreamSerializable, new()
    {
        public TResponse Response;
        public DateTime ReceivedDate;
    }

    /// <summary>
    /// Consumer implementation. For now partitions handled by the consumer must be specified,
    /// the offset fetch/commit API is not supported.
    /// </summary>
    class ConsumeRouter : IConsumeRouter
    {
        #region Batching

        class ByNodeFetchBatchingStrategy : IBatchStrategy<FetchMessage>
        {
            public bool Send(INode node, FetchMessage message)
            {
                return node.Fetch(message);
            }

            public void Dispose()
            {
            }
        }

        class GlobalFetchBatchingStrategy : IBatchStrategy<FetchMessage>
        {
            private readonly AccumulatorByNodeByTopic<FetchMessage> _accumulator;
            private Action<INode, IBatchByTopic<FetchMessage>> _batch;

            public GlobalFetchBatchingStrategy(Configuration configuration, Action<INode, IBatchByTopic<FetchMessage>> batch)
            {
                _accumulator = new AccumulatorByNodeByTopic<FetchMessage>(
                    m => m.Topic,
                    configuration.ConsumeBatchSize,
                    configuration.ConsumeBufferingTime);
                _accumulator.NewBatch += batch;
            }

            public bool Send(INode node, FetchMessage message)
            {
                return _accumulator.Add(Tuple.Create(node, message));
            }

            public void Dispose()
            {
                _accumulator.Dispose();
                _accumulator.NewBatch -= _batch;
                _batch = null;
            }
        }

        class ByNodeOffsetBatchingStrategy : IBatchStrategy<OffsetMessage>
        {
            public bool Send(INode node, OffsetMessage message)
            {
                return node.Offset(message);
            }

            public void Dispose()
            {
            }
        }

        class GlobalOffsetBatchingStrategy : IBatchStrategy<OffsetMessage>
        {
            private readonly AccumulatorByNodeByTopic<OffsetMessage> _accumulator;
            private Action<INode, IBatchByTopic<OffsetMessage>> _batch;

            public GlobalOffsetBatchingStrategy(Configuration configuration, Action<INode, IBatchByTopic<OffsetMessage>> batch)
            {
                _accumulator = new AccumulatorByNodeByTopic<OffsetMessage>(
                    m => m.Topic,
                    configuration.ConsumeBatchSize,
                    configuration.ConsumeBufferingTime);
                _accumulator.NewBatch += batch;
            }

            public bool Send(INode node, OffsetMessage message)
            {
                return _accumulator.Add(Tuple.Create(node, message));
            }

            public void Dispose()
            {
                _accumulator.Dispose();
                _accumulator.NewBatch -= _batch;
                _batch = null;
            }
        }

        #endregion

        #region Messages

        private enum ConsumerMessageType
        {
            FetchResponse,
            OffsetResponse,
            CheckPostponed,
            CheckPostponedAfterRoutingTableChange,
            Start,
            Stop,
            StartSubscription,
            Commit,
            Heartbeat,
        }

        private struct ConsumerMessage
        {
            public ConsumerMessageType MessageType;
            public ConsumerMessageValue MessageValue;
        }

        class TopicOnOff
        {
            public string Topic;
            public long Offset;
            public int Partition;
        }

        class CommitMsg
        {
            public TaskCompletionSource<bool> CommitPromise;
            public string Topic;
            public int Partition;
            public long Offset;
        }

        [StructLayout(LayoutKind.Explicit)]
        struct ConsumerMessageValue
        {
            [FieldOffset(0)]
            public CommonAcknowledgement<FetchResponse> FetchResponse;

            [FieldOffset(0)]
            public CommonAcknowledgement<CommonResponse<OffsetPartitionResponse>> OffsetResponse;

            [FieldOffset(0)]
            public IEnumerable<string> Subscription;

            [FieldOffset(0)]
            public TopicOnOff Topic;

            [FieldOffset(0)]
            public CommitMsg CommitMsg;
        }

{'fraction_non_alphanumeric': 0.08568945008692469, 'fraction_numerical': 0.0025619910330313845, 'mean_word_length': 1.9129797441364607, 'pattern_counts': {'":': 0, '<': 183, '<?xml version=': 0, '>': 187, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1685077', 'n_tokens_mistral': 5344, 'n_tokens_neox': 5088, 'n_words': 1788}