static void handleWriteToChunk ( ByteChannel sock , AutoBuffer ab ) throws IOException { String frameKey = ab . getStr ( ) ; byte [ ] expectedTypes = ab . getA1 ( ) ; if ( expectedTypes == null ) { throw new RuntimeException ( "Expected types can't be null." ) ; } int [ ] maxVecSizes = ab . getA4 ( ) ; int [ ] elemSizes = ExternalFrameUtils . getElemSizes ( expectedTypes , maxVecSizes != null ? maxVecSizes : EMPTY_ARI ) ; int [ ] startPos = ExternalFrameUtils . getStartPositions ( elemSizes ) ; byte [ ] vecTypes = vecTypesFromExpectedTypes ( expectedTypes , maxVecSizes != null ? maxVecSizes : EMPTY_ARI ) ; int expectedNumRows = ab . getInt ( ) ; int currentRowIdx = 0 ; int chunk_id = ab . getInt ( ) ; NewChunk [ ] nchnk = ChunkUtils . createNewChunks ( frameKey , vecTypes , chunk_id ) ; assert nchnk != null ; while ( currentRowIdx < expectedNumRows ) { for ( int typeIdx = 0 ; typeIdx < expectedTypes . length ; typeIdx ++ ) { switch ( expectedTypes [ typeIdx ] ) { case EXPECTED_BOOL : // fall through to byte since BOOL is internally stored in frame as number (byte) case EXPECTED_BYTE : store ( ab , nchnk [ startPos [ typeIdx ] ] , ab . get1 ( ) ) ; break ; case EXPECTED_CHAR : store ( ab , nchnk [ startPos [ typeIdx ] ] , ab . get2 ( ) ) ; break ; case EXPECTED_SHORT : store ( ab , nchnk [ startPos [ typeIdx ] ] , ab . get2s ( ) ) ; break ; case EXPECTED_INT : store ( ab , nchnk [ startPos [ typeIdx ] ] , ab . getInt ( ) ) ; break ; case EXPECTED_TIMESTAMP : // fall through to long since TIMESTAMP is internally stored in frame as long case EXPECTED_LONG : store ( ab , nchnk [ startPos [ typeIdx ] ] , ab . get8 ( ) ) ; break ; case EXPECTED_FLOAT : store ( nchnk [ startPos [ typeIdx ] ] , ab . get4f ( ) ) ; break ; case EXPECTED_DOUBLE : store ( nchnk [ startPos [ typeIdx ] ] , ab . get8d ( ) ) ; break ; case EXPECTED_STRING : store ( ab , nchnk [ startPos [ typeIdx ] ] , ab . getStr ( ) ) ; break ; case EXPECTED_VECTOR : storeVector ( ab , nchnk , elemSizes [ typeIdx ] , startPos [ typeIdx ] ) ; break ; default : throw new IllegalArgumentException ( "Unknown expected type: " + expectedTypes [ typeIdx ] ) ; } } currentRowIdx ++ ; } // close chunks at the end ChunkUtils . closeNewChunks ( nchnk ) ; // Flag informing sender that all work is done and // chunks are ready to be finalized. // // This also needs to be sent because in the sender we have to // wait for all chunks to be written to DKV; otherwise we get race during finalizing and // it happens that we try to finalize frame with chunks not ready yet AutoBuffer outputAb = new AutoBuffer ( ) ; outputAb . put1 ( ExternalFrameHandler . CONFIRM_WRITING_DONE ) ; writeToChannel ( outputAb , sock ) ; }
Internal method use on the h2o backend side to handle writing to the chunk from non - h2o environment