<commit_before>/*
 * Copyright 2014-2018 Real Logic Ltd.
 *
 * 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
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#ifndef AERON_IMAGE_H
#define AERON_IMAGE_H

#include <concurrent/AtomicBuffer.h>
#include <concurrent/logbuffer/LogBufferDescriptor.h>
#include <concurrent/logbuffer/FrameDescriptor.h>
#include <concurrent/logbuffer/Header.h>
#include <concurrent/logbuffer/TermReader.h>
#include <concurrent/logbuffer/TermBlockScanner.h>
#include <concurrent/status/UnsafeBufferPosition.h>
#include <algorithm>
#include <array>
#include <atomic>
#include <cassert>
#include "LogBuffers.h"

namespace aeron {

using namespace aeron::concurrent;
using namespace aeron::concurrent::logbuffer;
using namespace aeron::concurrent::status;

static UnsafeBufferPosition NULL_POSITION;

enum class ControlledPollAction : int
{
    /**
     * Abort the current polling operation and do not advance the position for this fragment.
     */
    ABORT = 1,

    /**
     * Break from the current polling operation and commit the position as of the end of the current fragment
     * being handled.
     */
    BREAK,

    /**
     * Continue processing but commit the position as of the end of the current fragment so that
     * flow control is applied to this point.
     */
    COMMIT,

    /**
     * Continue processing taking the same approach as the in fragment_handler_t.
     */
    CONTINUE
};

/**
 * Callback for handling fragments of data being read from a log.
 *
 * @param buffer containing the data.
 * @param offset at which the data begins.
 * @param length of the data in bytes.
 * @param header representing the meta data for the data.
 * @return The action to be taken with regard to the stream position after the callback.
 */
typedef std::function<ControlledPollAction(
    concurrent::AtomicBuffer &buffer,
    util::index_t offset,
    util::index_t length,
    Header &header)> controlled_poll_fragment_handler_t;

/**
 * Represents a replicated publication {@link Image} from a publisher to a {@link Subscription}.
 * Each {@link Image} identifies a source publisher by session id.
 *
 * Is an overlay on the LogBuffers and Position. So, can be effectively copied and moved.
 */
class Image
{
public:
    typedef Image this_t;

    Image() :
        m_header(0, 0, this),
        m_subscriberPosition(NULL_POSITION)
    {
    }

    /**
     * Construct a new image over a log to represent a stream of messages from a {@link Publication}.
     *
     * @param sessionId                  of the stream of messages.
     * @param initialPosition            at which the subscriber is joining the stream.
     * @param subscriberPosition         for indicating the position of the subscriber in the stream.
     * @param logBuffers                 containing the stream of messages.
     * @param correlationId              of the image with the media driver.
     * @param subscriptionRegistrationId of the Subscription.
     * @param exceptionHandler           to call if an exception is encountered on polling.
     */
    Image(
        std::int32_t sessionId,
        std::int64_t correlationId,
        std::int64_t subscriptionRegistrationId,
        const std::string& sourceIdentity,
        UnsafeBufferPosition& subscriberPosition,
        std::shared_ptr<LogBuffers> logBuffers,
        const exception_handler_t& exceptionHandler) :
        m_header(
            LogBufferDescriptor::initialTermId(logBuffers->atomicBuffer(LogBufferDescriptor::LOG_META_DATA_SECTION_INDEX)),
            logBuffers->atomicBuffer(0).capacity(),
            this),
        m_subscriberPosition(subscriberPosition),
        m_logBuffers(logBuffers),
        m_sourceIdentity(sourceIdentity),
        m_isClosed(false),
        m_exceptionHandler(exceptionHandler),
        m_correlationId(correlationId),
        m_subscriptionRegistrationId(subscriptionRegistrationId),
        m_sessionId(sessionId)
    {
        for (int i = 0; i < LogBufferDescriptor::PARTITION_COUNT; i++)
        {
            m_termBuffers[i] = logBuffers->atomicBuffer(i);
        }

        const util::index_t capacity = m_termBuffers[0].capacity();

        m_joinPosition = subscriberPosition.get();
        m_finalPosition = m_joinPosition;
        m_termLengthMask = capacity - 1;
        m_positionBitsToShift = BitUtil::numberOfTrailingZeroes(capacity);
        m_isEos = false;
    }

    Image(const Image& image) :
        m_termBuffers(image.m_termBuffers),
        m_header(image.m_header),
        m_subscriberPosition(image.m_subscriberPosition),
        m_logBuffers(image.m_logBuffers),
        m_sourceIdentity(image.m_sourceIdentity),
        m_isClosed(image.isClosed()),
        m_exceptionHandler(image.m_exceptionHandler),
        m_correlationId(image.m_correlationId),
        m_subscriptionRegistrationId(image.m_subscriptionRegistrationId),
        m_joinPosition(image.m_joinPosition),
        m_finalPosition(image.m_finalPosition),
        m_sessionId(image.m_sessionId),
        m_termLengthMask(image.m_termLengthMask),
        m_positionBitsToShift(image.m_positionBitsToShift),
        m_isEos(image.m_isEos)
    {
        m_subscriberPosition.wrap(image.m_subscriberPosition);
    }

    Image& operator=(const Image& image)
    {
        m_termBuffers = image.m_termBuffers;
        m_header = image.m_header;
        m_subscriberPosition.wrap(image.m_subscriberPosition);<commit_msg>[C++] Position.wrap is equivalent to copying.<commit_after>/*
 * Copyright 2014-2018 Real Logic Ltd.
 *
 * 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
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#ifndef AERON_IMAGE_H
#define AERON_IMAGE_H

#include <concurrent/AtomicBuffer.h>
#include <concurrent/logbuffer/LogBufferDescriptor.h>
#include <concurrent/logbuffer/FrameDescriptor.h>
#include <concurrent/logbuffer/Header.h>
#include <concurrent/logbuffer/TermReader.h>
#include <concurrent/logbuffer/TermBlockScanner.h>
#include <concurrent/status/UnsafeBufferPosition.h>
#include <algorithm>
#include <array>
#include <atomic>
#include <cassert>
#include "LogBuffers.h"

namespace aeron {

using namespace aeron::concurrent;
using namespace aeron::concurrent::logbuffer;
using namespace aeron::concurrent::status;

static UnsafeBufferPosition NULL_POSITION;

enum class ControlledPollAction : int
{
    /**
     * Abort the current polling operation and do not advance the position for this fragment.
     */
    ABORT = 1,

    /**
     * Break from the current polling operation and commit the position as of the end of the current fragment
     * being handled.
     */
    BREAK,

    /**
     * Continue processing but commit the position as of the end of the current fragment so that
     * flow control is applied to this point.
     */
    COMMIT,

    /**
     * Continue processing taking the same approach as the in fragment_handler_t.
     */
    CONTINUE
};

/**
 * Callback for handling fragments of data being read from a log.
 *
 * @param buffer containing the data.
 * @param offset at which the data begins.
 * @param length of the data in bytes.
 * @param header representing the meta data for the data.
 * @return The action to be taken with regard to the stream position after the callback.
 */
typedef std::function<ControlledPollAction(
    concurrent::AtomicBuffer &buffer,
    util::index_t offset,
    util::index_t length,
    Header &header)> controlled_poll_fragment_handler_t;

/**
 * Represents a replicated publication {@link Image} from a publisher to a {@link Subscription}.
 * Each {@link Image} identifies a source publisher by session id.
 *
 * Is an overlay on the LogBuffers and Position. So, can be effectively copied and moved.
 */
class Image
{
public:
    typedef Image this_t;

    Image() :
        m_header(0, 0, this),
        m_subscriberPosition(NULL_POSITION)
    {
    }

    /**
     * Construct a new image over a log to represent a stream of messages from a {@link Publication}.
     *
     * @param sessionId                  of the stream of messages.
     * @param initialPosition            at which the subscriber is joining the stream.
     * @param subscriberPosition         for indicating the position of the subscriber in the stream.
     * @param logBuffers                 containing the stream of messages.
     * @param correlationId              of the image with the media driver.
     * @param subscriptionRegistrationId of the Subscription.
     * @param exceptionHandler           to call if an exception is encountered on polling.
     */
    Image(
        std::int32_t sessionId,
        std::int64_t correlationId,
        std::int64_t subscriptionRegistrationId,
        const std::string& sourceIdentity,
        UnsafeBufferPosition& subscriberPosition,
        std::shared_ptr<LogBuffers> logBuffers,
        const exception_handler_t& exceptionHandler) :
        m_header(
            LogBufferDescriptor::initialTermId(logBuffers->atomicBuffer(LogBufferDescriptor::LOG_META_DATA_SECTION_INDEX)),
            logBuffers->atomicBuffer(0).capacity(),
            this),
        m_subscriberPosition(subscriberPosition),
        m_logBuffers(logBuffers),
        m_sourceIdentity(sourceIdentity),
        m_isClosed(false),
        m_exceptionHandler(exceptionHandler),
        m_correlationId(correlationId),
        m_subscriptionRegistrationId(subscriptionRegistrationId),
        m_sessionId(sessionId)
    {
        for (int i = 0; i < LogBufferDescriptor::PARTITION_COUNT; i++)
        {
            m_termBuffers[i] = logBuffers->atomicBuffer(i);
        }

        const util::index_t capacity = m_termBuffers[0].capacity();

        m_joinPosition = subscriberPosition.get();
        m_finalPosition = m_joinPosition;
        m_termLengthMask = capacity - 1;
        m_positionBitsToShift = BitUtil::numberOfTrailingZeroes(capacity);
        m_isEos = false;
    }

    Image(const Image& image) :
        m_termBuffers(image.m_termBuffers),
        m_header(image.m_header),
        m_subscriberPosition(image.m_subscriberPosition),
        m_logBuffers(image.m_logBuffers),
        m_sourceIdentity(image.m_sourceIdentity),
        m_isClosed(image.isClosed()),
        m_exceptionHandler(image.m_exceptionHandler),
        m_correlationId(image.m_correlationId),
        m_subscriptionRegistrationId(image.m_subscriptionRegistrationId),
        m_joinPosition(image.m_joinPosition),
        m_finalPosition(image.m_finalPosition),
        m_sessionId(image.m_sessionId),
        m_termLengthMask(image.m_termLengthMask),
        m_positionBitsToShift(image.m_positionBitsToShift),
        m_isEos(image.m_isEos)
    {
    }

    Image& operator=(const Image& image)
    {
        m_termBuffers = image.m_termBuffers;
        m_header = image.m_header;
        m_subscriberPosition = image.m_subscriberPosition;
{'fraction_non_alphanumeric': 0.08083297917552061, 'fraction_numerical': 0.004249893752656183, 'mean_word_length': 3.223259152907394, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 35, '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': '2469834', 'n_tokens_mistral': 3287, 'n_tokens_neox': 3188, 'n_words': 954}