class ImageRotationStage extends PronghornStage {

    @Override
    public void run() {

        // Read in a next frame if we do not have one.
        if (nextInputFrameTimestamp == NONE && PipeReader.tryReadFragment(imagePipeIn)) {
            assert PipeReader.getMsgIdx(imagePipeIn) == ImageSchema.MSG_FRAMESTART_1;
            nextInputFrameWidth = PipeReader.readInt(imagePipeIn, ImageSchema.MSG_FRAMESTART_1_FIELD_WIDTH_101);
            nextInputFrameHeight = PipeReader.readInt(imagePipeIn, ImageSchema.MSG_FRAMESTART_1_FIELD_HEIGHT_201);
            nextInputFrameTimestamp = PipeReader.readLong(imagePipeIn, ImageSchema.MSG_FRAMESTART_1_FIELD_TIMESTAMP_301);
            nextInputFrameSize = PipeReader.readInt(imagePipeIn, ImageSchema.MSG_FRAMESTART_1_FIELD_FRAMEBYTES_401);
            assert inputFrameHead == NONE || inputFrameHead == FRAME_BUFFERED;
            inputFrameHead = 0;

            PipeReader.releaseReadLock(imagePipeIn);
        }

        // Read in a cropping request if we do not have one.
        if (rotationRequestTimestamp == NONE && PipeReader.tryReadFragment(imageRotationPipeIn)) {
            assert PipeReader.getMsgIdx(imageRotationPipeIn) == ImageRotationSchema.MSG_ROTATIONREQUEST_1;

            // Read in rotation data.
            long rotationRequestNumerator = PipeReader.readLong(imageRotationPipeIn, ImageRotationSchema.MSG_ROTATIONREQUEST_1_FIELD_NUMERATOR_101);
            long rotationRequestDenominator = PipeReader.readLong(imageRotationPipeIn, ImageRotationSchema.MSG_ROTATIONREQUEST_1_FIELD_DENOMINATOR_201);
            rotationRequestTimestamp = PipeReader.readLong(imageRotationPipeIn, ImageRotationSchema.MSG_ROTATIONREQUEST_1_FIELD_TIMESTAMP_301);

            // Calculate theta degrees from rotation data.
            double theta = rotationRequestNumerator / rotationRequestDenominator;

            // Calculate alpha, beta, and gamma for rotation data.
            rotationAlpha = -1 * Math.tan(theta / 2);
            rotationBeta = Math.sin(theta);

            PipeReader.releaseReadLock(imageRotationPipeIn);
        }

        // Read in the next frame if the next frame's timestamp is older than the rotation request.
        if (nextInputFrameTimestamp != NONE && nextInputFrameTimestamp < rotationRequestTimestamp) {

            // Ensure image frame buffer is correctly sized.
            if (inputFrame == null || inputFrame.length != nextInputFrameSize) {
                inputFrame = new byte[nextInputFrameSize];
            }

            // Fill frame arrays.
            while (PipeReader.tryReadFragment(imagePipeIn)) {
                assert PipeReader.getMsgIdx(imagePipeIn) == ImageSchema.MSG_FRAMECHUNK_2;

                // Read data.
                inputFrameHead += PipeReader.readBytes(imagePipeIn, ImageSchema.MSG_FRAMECHUNK_2_FIELD_ROWBYTES_102, inputFrame, inputFrameHead);

                // If frame is full, we're done!
                if (inputFrameHead >= inputFrame.length) {
                    inputFrameTimestamp = nextInputFrameTimestamp;
                    inputFrameWidth = nextInputFrameWidth;
                    inputFrameHeight = nextInputFrameHeight;
                    inputFrameHead = FRAME_BUFFERED;
                    nextInputFrameTimestamp = NONE;
                    break;
                }

                PipeReader.releaseReadLock(imagePipeIn);
            }
        }

        // If we have an input frame buffered, have an active rotation request,
        // and the input frame predates the rotation request, write out a rotation.
        if (outputFrameHead != NONE ||
            (inputFrameHead == FRAME_BUFFERED && rotationRequestTimestamp != NONE && inputFrameTimestamp < rotationRequestTimestamp)) {

            // Frame start.
            if (outputFrameHead == NONE && PipeWriter.tryWriteFragment(imagePipeOut, ImageSchema.MSG_FRAMESTART_1)) {

                // Write start data to the pipe.
                PipeWriter.writeInt(imagePipeOut, ImageSchema.MSG_FRAMESTART_1_FIELD_WIDTH_101, cropWidth);
                PipeWriter.writeInt(imagePipeOut, ImageSchema.MSG_FRAMESTART_1_FIELD_HEIGHT_201, cropHeight);
                PipeWriter.writeLong(imagePipeOut, ImageSchema.MSG_FRAMESTART_1_FIELD_TIMESTAMP_301, System.currentTimeMillis());
                PipeWriter.writeInt(imagePipeOut, ImageSchema.MSG_FRAMESTART_1_FIELD_FRAMEBYTES_401, outputFrame.length);
                PipeWriter.writeInt(imagePipeOut, ImageSchema.MSG_FRAMESTART_1_FIELD_BITSPERPIXEL_501, 24);
                PipeWriter.writeBytes(imagePipeOut, ImageSchema.MSG_FRAMESTART_1_FIELD_ENCODING_601, LinuxImageCaptureStage.OUTPUT_ENCODING);
                PipeWriter.publishWrites(imagePipeOut);
                outputFrameHead = 0;

                // Reset the output frame.
                // This must be done because some positions in the frame may not be written to,
                // and thus should appear empty (black).
                Arrays.fill(outputFrame, (byte) 0x00);

                // Skew the input frame into our output frame.
                shearRotateFrame(rotationAlpha, rotationBeta, inputFrame, inputFrameWidth, inputFrameHeight, outputFrame, cropWidth, cropHeight);
            }

            // Write output frame data.
            if (outputFrameHead >= 0 && PipeWriter.tryWriteFragment(imagePipeOut, ImageSchema.MSG_FRAMECHUNK_2)) {

                // Write bytes.
                PipeWriter.writeBytes(imagePipeOut, ImageSchema.MSG_FRAMECHUNK_2_FIELD_ROWBYTES_102, outputFrame, outputFrameHead, outputFrameRowSize);
                PipeWriter.publishWrites(imagePipeOut);

                // Progress head.
                outputFrameHead += outputFrameRowSize;

                // If the head exceeds the size of the frame bytes, we're done writing.
                if (outputFrameHead >= outputFrame.length) {
                    outputFrameHead = NONE;
                }
            }
        }
    }

    public  ImageRotationStage(GraphManager gm,
                              Pipe<ImageSchema> imagePipeIn,
                              Pipe<ImageRotationSchema> imageRotationPipeIn,
                              Pipe<ImageSchema> imagePipeOut);
    public  ImageRotationStage(GraphManager gm,
                              Pipe<ImageSchema> imagePipeIn,
                              Pipe<ImageRotationSchema> imageRotationPipeIn,
                              Pipe<ImageSchema> imagePipeOut,
                              int cropWidth, int cropHeight);

}