BlobDeserializer implements EventDeserializer { @SuppressWarnings("resource") @Override public Event readEvent() throws IOException { ensureOpen(); ByteArrayOutputStream blob = null; byte[] buf = new byte[Math.min(maxBlobLength, DEFAULT_BUFFER_SIZE)]; int blobLength = 0; int n = 0; while ((n = in.read(buf, 0, Math.min(buf.length, maxBlobLength - blobLength))) != -1) { if (blob == null) { blob = new ByteArrayOutputStream(n); } blob.write(buf, 0, n); blobLength += n; if (blobLength >= maxBlobLength) { LOGGER.warn("File length exceeds maxBlobLength ({}), truncating BLOB event!", maxBlobLength); break; } } if (blob == null) { return null; } else { return EventBuilder.withBody(blob.toByteArray()); } } protected  BlobDeserializer(Context context, ResettableInputStream in); @SuppressWarnings("resource") @Override Event readEvent(); @Override List<Event> readEvents(int numEvents); @Override void mark(); @Override void reset(); @Override void close(); static final String MAX_BLOB_LENGTH_KEY; static final int MAX_BLOB_LENGTH_DEFAULT; }
@Test public void testMaxLineLength() throws IOException { String longLine = "abcdefghijklmnopqrstuvwxyz\n"; Context ctx = new Context(); ctx.put(BlobDeserializer.MAX_BLOB_LENGTH_KEY, "10"); ResettableInputStream in = new ResettableTestStringInputStream(longLine); EventDeserializer des = new BlobDeserializer(ctx, in); assertEventBodyEquals("abcdefghij", des.readEvent()); assertEventBodyEquals("klmnopqrst", des.readEvent()); assertEventBodyEquals("uvwxyz\n", des.readEvent()); assertNull(des.readEvent()); }