@@ -0,0 +1,3170 @@
+/**%0A * Copyright 2015 Netflix, Inc.%0A *%0A * Licensed under the Apache License, Version 2.0 (the %22License%22);%0A * you may not use this file except in compliance with the License.%0A * You may obtain a copy of the License at%0A *%0A * http://www.apache.org/licenses/LICENSE-2.0%0A *%0A * Unless required by applicable law or agreed to in writing, software%0A * distributed under the License is distributed on an %22AS IS%22 BASIS,%0A * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A * See the License for the specific language governing permissions and%0A * limitations under the License.%0A */%0Apackage io.reactivesocket.internal;%0A%0Aimport io.reactivesocket.Frame;%0Aimport uk.co.real_logic.agrona.MutableDirectBuffer;%0Aimport uk.co.real_logic.agrona.concurrent.OneToOneConcurrentArrayQueue;%0Aimport uk.co.real_logic.agrona.concurrent.UnsafeBuffer;%0A%0Aimport java.nio.ByteBuffer;%0A%0Apublic class ThreadSafeFramePool implements FramePool%0A%7B%0A    private static final int MAX_CACHED_FRAMES = 16;%0A%0A    private static final OneToOneConcurrentArrayQueue%3CFrame%3E FRAME_QUEUE =%0A        new OneToOneConcurrentArrayQueue%3C%3E(MAX_CACHED_FRAMES);%0A%0A    private static final OneToOneConcurrentArrayQueue%3CMutableDirectBuffer%3E DIRECTBUFFER_QUEUE =%0A        new OneToOneConcurrentArrayQueue%3C%3E(MAX_CACHED_FRAMES);%0A%0A    public Frame acquireFrame(int size)%0A    %7B%0A        final MutableDirectBuffer directBuffer = acquireMutableDirectBuffer(size);%0A%0A        Frame frame = pollFrame();%0A        if (null == frame)%0A        %7B%0A            frame = Frame.allocate(directBuffer);%0A        %7D%0A%0A        return frame;%0A    %7D%0A%0A    public Frame acquireFrame(ByteBuffer byteBuffer)%0A    %7B%0A        return Frame.allocate(new UnsafeBuffer(byteBuffer));%0A    %7D%0A%0A    public Frame acquireFrame(MutableDirectBuffer mutableDirectBuffer)%0A    %7B%0A        Frame frame = pollFrame();%0A        if (null == frame)%0A        %7B%0A            frame = Frame.allocate(mutableDirectBuffer);%0A        %7D%0A%0A        return frame;%0A    %7D%0A%0A    public MutableDirectBuffer acquireMutableDirectBuffer(ByteBuffer byteBuffer)%0A    %7B%0A        MutableDirectBuffer directBuffer = pollMutableDirectBuffer();%0A        if (null == directBuffer)%0A        %7B%0A            directBuffer = new UnsafeBuffer(byteBuffer);%0A        %7D%0A%0A        return directBuffer;%0A    %7D%0A%0A    public MutableDirectBuffer acquireMutableDirectBuffer(int size)%0A    %7B%0A        UnsafeBuffer directBuffer = (UnsafeBuffer)pollMutableDirectBuffer();%0A        if (null == directBuffer %7C%7C directBuffer.byteBuffer().capacity() %3C size)%0A        %7B%0A            directBuffer = new UnsafeBuffer(ByteBuffer.allocate(size));%0A        %7D%0A        else%0A        %7B%0A            directBuffer.byteBuffer().limit(size).position(0);%0A        %7D%0A%0A        return directBuffer;%0A    %7D%0A%0A    public synchronized void release(Frame frame)%0A    %7B%0A        FRAME_QUEUE.offer(frame);%0A    %7D%0A%0A    public synchronized void release(MutableDirectBuffer mutableDirectBuffer)%0A    %7B%0A        DIRECTBUFFER_QUEUE.offer(mutableDirectBuffer);%0A    %7D%0A%0A    private synchronized Frame pollFrame()%0A    %7B%0A        return FRAME_QUEUE.poll();%0A    %7D%0A%0A    private synchronized MutableDirectBuffer pollMutableDirectBuffer()%0A    %7B%0A        return DIRECTBUFFER_QUEUE.poll();%0A    %7D%0A%7D%0A

add thread safe frame pool