StreamSource implements Serializable { public boolean validate ( ) { if (isValidated==true) return validationResult; windowingType = DEFAULT_WINDOW_TYPE; isValidated=true; if (samplingRate <=0 ) logger.warn( new StringBuilder( ).append( "The sampling rate is set to zero (or negative) which means no results. StreamSource = " ) .append( getAlias( ) ).toString( ) ); else if(samplingRate > 1){ samplingRate = 1; logger.warn( new StringBuilder( ).append( "The provided sampling rate is greater than 1, resetting it to 1. StreamSource = " ) .append( getAlias( ) ).toString( ) ); } if (getAddressing().length==0) { logger.warn(new StringBuilder("Validation failed because there is no addressing predicates provided for the stream source (the addressing part of the stream source is empty)").append("stream source alias = ").append(getAlias())); return validationResult=false; } if ( this.rawHistorySize != null ) { this.rawHistorySize = this.rawHistorySize.replace( " " , "" ).trim( ).toLowerCase( ); if ( this.rawHistorySize.equalsIgnoreCase( "" ) ) return validationResult = true; final int second = 1000; final int minute = second * 60; final int hour = minute * 60; final int mIndex = this.rawHistorySize.indexOf( "m" ); final int hIndex = this.rawHistorySize.indexOf( "h" ); final int sIndex = this.rawHistorySize.indexOf( "s" ); if ( mIndex < 0 && hIndex < 0 && sIndex < 0 ) { try { this.parsedStorageSize = Long.parseLong(this.rawHistorySize ); this.isStorageCountBased = true; windowingType = WindowType.TUPLE_BASED; } catch ( final NumberFormatException e ) { logger.error( new StringBuilder( ).append( "The storage size, " ).append( this.rawHistorySize ).append( ", specified for the Stream Source : " ).append( this.getAlias( ) ).append( " is not valid." ).toString( ) , e ); return (validationResult= false); } } else try { final StringBuilder shs = new StringBuilder( this.rawHistorySize ); if ( mIndex >= 0 && mIndex == shs.length() - 1) this.parsedStorageSize = Long.parseLong(shs.deleteCharAt( mIndex ).toString( ) ) * minute; else if ( hIndex >= 0 && hIndex == shs.length() - 1) this.parsedStorageSize = Long.parseLong( shs.deleteCharAt( hIndex ).toString( ) ) * hour; else if ( sIndex >= 0 && sIndex == shs.length() - 1) this.parsedStorageSize = Long.parseLong( shs.deleteCharAt( sIndex ).toString( ) ) * second; else Long.parseLong(""); this.isStorageCountBased = false; windowingType = WindowType.TIME_BASED; } catch ( NumberFormatException e ) { logger.debug( e.getMessage( ) , e ); logger.error( new StringBuilder( ).append( "The storage size, " ).append( this.rawHistorySize ).append( ", specified for the Stream Source : " ).append( this.getAlias( ) ).append( " is not valid." ).toString( ) ); return (validationResult=false); } } logger.debug("validate() called"); if(this.rawSlideValue == null){ windowingType = (windowingType == WindowType.TUPLE_BASED) ? WindowType.TUPLE_BASED_SLIDE_ON_EACH_TUPLE : WindowType.TIME_BASED_SLIDE_ON_EACH_TUPLE; return validationResult = true; } else { this.rawSlideValue = this.rawSlideValue.replace( " " , "" ).trim( ).toLowerCase( ); if ( this.rawSlideValue.equalsIgnoreCase( "" ) ){ windowingType = (windowingType == WindowType.TUPLE_BASED) ? WindowType.TUPLE_BASED_SLIDE_ON_EACH_TUPLE : WindowType.TIME_BASED_SLIDE_ON_EACH_TUPLE; return validationResult = true; } final int second = 1000; final int minute = second * 60; final int hour = minute * 60; final int mIndex = this.rawSlideValue.indexOf( "m" ); final int hIndex = this.rawSlideValue.indexOf( "h" ); final int sIndex = this.rawSlideValue.indexOf( "s" ); if ( mIndex < 0 && hIndex < 0 && sIndex < 0 ) { try { this.parsedSlideValue = Long.parseLong( this.rawSlideValue ); if(parsedSlideValue == 1){ windowingType = (windowingType == WindowType.TIME_BASED) ? WindowType.TIME_BASED_SLIDE_ON_EACH_TUPLE : WindowType.TUPLE_BASED_SLIDE_ON_EACH_TUPLE; } else if(windowingType == WindowType.TIME_BASED) windowingType = WindowType.TIME_BASED_WIN_TUPLE_BASED_SLIDE; } catch ( final NumberFormatException e ) { logger.error( new StringBuilder( ).append( "The slide value, " ).append( this.rawSlideValue ).append( ", specified for the Stream Source : " ).append( this.getAlias( ) ).append( " is not valid." ).toString( ) , e ); return (validationResult= false); } } else try { final StringBuilder shs = new StringBuilder( this.rawSlideValue ); if ( mIndex >= 0 && mIndex == shs.length() - 1) this.parsedSlideValue = Long.parseLong( shs.deleteCharAt( mIndex ).toString( ) ) * minute; else if ( hIndex >= 0 && hIndex == shs.length() - 1) this.parsedSlideValue = Long.parseLong( shs.deleteCharAt( hIndex ).toString( ) ) * hour; else if ( sIndex >= 0 && sIndex == shs.length() - 1) this.parsedSlideValue = Long.parseLong( shs.deleteCharAt( sIndex ).toString( ) ) * second; else Long.parseLong(""); if(windowingType == WindowType.TUPLE_BASED) windowingType = WindowType.TUPLE_BASED_WIN_TIME_BASED_SLIDE; } catch ( NumberFormatException e ) { logger.debug( e.getMessage( ) , e ); logger.error( new StringBuilder( ).append( "The slide value, " ).append( this.rawSlideValue ).append( ", specified for the Stream Source : " ).append( this.getAlias( ) ).append( " is not valid." ).toString( ) ); return (validationResult=false); } } return validationResult=true; } StreamSource(); String getRawHistorySize(); StreamSource setRawHistorySize(String rawHistorySize); StreamSource setRawSlideValue(String rawSlideValue); StreamSource setAddressing(AddressBean[] addressing); StreamSource setAlias(String alias); StreamSource setSqlQuery(String sqlQuery); AddressBean [] getAddressing( ); CharSequence getAlias( ); InputStream getInputStream(); int getDisconnectedBufferSize( ); void setDisconnectedBufferSize(int disconnectedBufferSize); float getSamplingRate( ); void setSamplingRate(float newRate ); String getStorageSize( ); String getSlideValue(); String getSqlQuery( ); void setWrapper( AbstractWrapper wrapper  ); AbstractWrapper getWrapper( ); boolean validate( ); boolean isStorageCountBased( ); long getParsedStorageSize( ); WindowType getWindowingType(); long getParsedSlideValue(); boolean windowSlided(); void setQueryRewriter(QueryRewriter rewriter); QueryRewriter getQueryRewriter(); StringBuilder rewrite(String query); StringBuilder getUIDStr(); int hashCode(); StringBuilder toSql(); StreamSource setInputStream(InputStream is); AddressBean getActiveAddressBean(); String toString(); static final String DEFAULT_QUERY; static final AddressBean[] EMPTY_ADDRESS_BEAN; static final long STORAGE_SIZE_NOT_SET; static final long DEFAULT_SLIDE_VALUE; static final WindowType DEFAULT_WINDOW_TYPE; }
@Test public void testValidate() { InputStream is = new InputStream(); StreamSource ss = new StreamSource().setAlias("my-stream").setAddressing(addressing).setSqlQuery("select * from wrapper").setRawHistorySize("10m"); assertTrue(ss.validate()); assertFalse(ss.isStorageCountBased()); assertEquals(ss.getParsedStorageSize(),10*60*1000); ss = new StreamSource().setAlias("my-stream").setAddressing(addressing).setSqlQuery("select * from wrapper").setRawHistorySize("10 m").setInputStream(is); assertFalse(ss.isStorageCountBased()); assertEquals(ss.getParsedStorageSize(),10*60*1000); ss = new StreamSource().setAlias("my-stream").setAddressing(addressing).setSqlQuery("select * from wrapper").setRawHistorySize("10 s").setInputStream(is); assertFalse(ss.isStorageCountBased()); assertEquals(ss.getParsedStorageSize(),10*1000); assertFalse(ss.isStorageCountBased()); ss = new StreamSource().setAlias("my-stream").setAddressing(addressing).setSqlQuery("select * from wrapper").setRawHistorySize("2 h").setInputStream(is); assertFalse(ss.isStorageCountBased()); assertEquals(ss.getParsedStorageSize(),2*60*60*1000); }