<commit_before>package org.cometd.server.ext;

import java.util.Map;

import org.cometd.bayeux.Channel;
import org.cometd.bayeux.server.ServerMessage;
import org.cometd.bayeux.server.ServerSession;
import org.cometd.bayeux.server.ServerMessage.Mutable;
import org.cometd.bayeux.server.ServerSession.Extension;
import org.eclipse.jetty.util.ArrayQueue;
import org.eclipse.jetty.util.log.Log;

/**
 * Acknowledged Message Client extension.
 * 
 * Tracks the batch id of messages sent to a client.
 * 
 */
public class AcknowledgedMessagesClientExtension implements Extension
{
    private final ServerSession _session;
    private final ArrayQueue<ServerMessage> _queue;
    private final ArrayIdQueue<ServerMessage> _unackedQueue;
    private long _lastAck;

    public AcknowledgedMessagesClientExtension(ServerSession session)
    {
        _session=session;
        _queue=(ArrayQueue<ServerMessage>)session.getQueue();
        _unackedQueue=new ArrayIdQueue<ServerMessage>(16,32,session.getQueue());
        _unackedQueue.setCurrentId(1);
    }

    /* ------------------------------------------------------------ */
    public boolean rcv(ServerSession from, Mutable message)
    {
        return true;
    }

    /* ------------------------------------------------------------ */
    public boolean rcvMeta(ServerSession session, Mutable message)
    {
        if (Channel.META_CONNECT.equals(message.getChannel()))
        {
            Map<String,Object> ext=message.getExt(false);
            if (ext != null)
            {
                assert session==_session;

                synchronized(_queue)
                {
                    Long acked=(Long)ext.get("ack");
                    if (acked != null)
                    {                        
                        if (acked==_lastAck)
                        {
                            Log.debug("double ACK");
                            _queue.clear();
                            _queue.addAll(_unackedQueue);
                        }
                        _lastAck=acked;
                        
                        // We have received an ack ID, so delete the acked
                        // messages.
                        final int s=_unackedQueue.size();
                        if (s > 0)
                        {
                            if (_unackedQueue.getAssociatedIdUnsafe(s - 1) <= acked)
                            {
                                // we can just clear the queue
                                for (int i=0; i < s; i++)
                                {
                                    final ServerMessage q=_unackedQueue.getUnsafe(i);
                                    q.decRef();
                                }
                                _unackedQueue.clear();
                            }
                            else
                            {
                                // we need to remove elements until we see unacked
                                for (int i=0; i < s; i++)
                                {
                                    final long a=_unackedQueue.getAssociatedIdUnsafe(0);
                                    if (a <= acked)
                                    {
                                        final ServerMessage q=_unackedQueue.remove();
                                        q.decRef();
                                        continue;
                                    }
                                    break;
                                }
                            }
                        }
                    }
                }

                // requeue all unacked messages.
                final int cid=_unackedQueue.getCurrentId();
                final int s=_unackedQueue.size();
                for (int i=0; i < s; i++)
                {
                    final long a=_unackedQueue.getAssociatedIdUnsafe(0);
                    if (a < cid)
                    {
                        final ServerMessage q=_unackedQueue.remove();
                        _queue.add(i,q);
                    }
                    else
                        break;
                }
            }
        }

        return true;
    }

    /* ------------------------------------------------------------ */
    public ServerMessage send(ServerSession to, ServerMessage message)
    {
        synchronized(_queue)
        {
            message.incRef();
            _unackedQueue.add(message);
        }
        
        return message;
    }

    /* ------------------------------------------------------------ */
    public boolean sendMeta(ServerSession to, Mutable message)
    {
        if (message.getChannel().equals(Channel.META_CONNECT))
        {
            synchronized(_queue)
            {
                Map<String,Object> ext=message.getExt(true);
                ext.put("ack",_unackedQueue.getCurrentId());
                _unackedQueue.incrementCurrentId();
            }
        }
        return true;
    }

}
    
    
<commit_msg>COMETD-76 dont hold client lock while sending<commit_after>package org.cometd.server.ext;

import java.util.Map;

import org.cometd.bayeux.Channel;
import org.cometd.bayeux.server.ServerMessage;
import org.cometd.bayeux.server.ServerSession;
import org.cometd.bayeux.server.ServerMessage.Mutable;
import org.cometd.bayeux.server.ServerSession.Extension;
import org.eclipse.jetty.util.ArrayQueue;
import org.eclipse.jetty.util.log.Log;

/**
 * Acknowledged Message Client extension.
 * 
 * Tracks the batch id of messages sent to a client.
 * 
 */
public class AcknowledgedMessagesClientExtension implements Extension
{
    private final ServerSession _session;
    private final ArrayQueue<ServerMessage> _queue;
    private final ArrayIdQueue<ServerMessage> _unackedQueue;
    private long _lastAck;

    public AcknowledgedMessagesClientExtension(ServerSession session)
    {
        _session=session;
        _queue=(ArrayQueue<ServerMessage>)session.getQueue();
        _unackedQueue=new ArrayIdQueue<ServerMessage>(16,32,session.getQueue());
        _unackedQueue.setCurrentId(1);
    }

    /* ------------------------------------------------------------ */
    public boolean rcv(ServerSession from, Mutable message)
    {
        return true;
    }

    /* ------------------------------------------------------------ */
    public boolean rcvMeta(ServerSession session, Mutable message)
    {
        if (Channel.META_CONNECT.equals(message.getChannel()))
        {
            Map<String,Object> ext=message.getExt(false);
            if (ext != null)
            {
                assert session==_session;

                synchronized(_queue)
                {
                    Long acked=(Long)ext.get("ack");
                    if (acked != null)
                    {                        
                        if (acked.longValue()<=_lastAck)
                        {
                            Log.debug(session+" lost ACK "+acked.longValue()+"<="+_lastAck);
                            _queue.clear();
                            _queue.addAll(_unackedQueue);
                        }
                        else
                        {
                            _lastAck=acked.longValue();

                            // We have received an ack ID, so delete the acked
                            // messages.
                            final int s=_unackedQueue.size();
                            if (s > 0)
                            {
                                if (_unackedQueue.getAssociatedIdUnsafe(s - 1) <= acked)
                                {
                                    // we can just clear the queue
                                    for (int i=0; i < s; i++)
                                    {
                                        final ServerMessage q=_unackedQueue.getUnsafe(i);
                                        q.decRef();
                                    }
                                    _unackedQueue.clear();
                                }
                                else
                                {
                                    // we need to remove elements until we see unacked
                                    for (int i=0; i < s; i++)
                                    {
                                        final long a=_unackedQueue.getAssociatedIdUnsafe(0);
                                        if (a <= acked)
                                        {
                                            final ServerMessage q=_unackedQueue.remove();
                                            q.decRef();
                                            continue;
                                        }
                                        break;
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }

        return true;
    }

    /* ------------------------------------------------------------ */
    public ServerMessage send(ServerSession to, ServerMessage message)
    {
        synchronized(_queue)
        {
            message.incRef();
            _unackedQueue.add(message);
        }
        
        return message;
    }

    /* ------------------------------------------------------------ */
    public boolean sendMeta(ServerSession to, Mutable message)
    {
        if (message.getChannel().equals(Channel.META_CONNECT))
        {
            synchronized(_queue)
            {
                Map<String,Object> ext=message.getExt(true);
                ext.put("ack",_unackedQueue.getCurrentId());
                _unackedQueue.incrementCurrentId();
            }
        }
        return true;
    }

}
    
    

{'fraction_non_alphanumeric': 0.12441836941128868, 'fraction_numerical': 0.002427675500708072, 'mean_word_length': 0.8991548213599693, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 44, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1379653', 'n_tokens_mistral': 2264, 'n_tokens_neox': 2173, 'n_words': 417}