public void on_data_available(DataReader reader){
		
		KeyedBytesDataReader datareader = (KeyedBytesDataReader) reader;
		SampleInfo info = new SampleInfo();
		
		while(true){
			try{
				KeyedBytes sample = new KeyedBytes();
				datareader.take_next_sample(sample,info);
				if(info.valid_data){
					@SuppressWarnings("unchecked")
					T msg = (T) ByteSerialzier.deserialize(sample.value);
					this.queue.put(msg);
				}
			}
			catch (InterruptedException e) {
				e.printStackTrace();
			}
			catch(RETCODE_NO_DATA noData){
				// no more data to read
				break;
			}
			catch(RETCODE_ERROR e){
				e.printStackTrace();
			} catch (ClassNotFoundException e) {
				e.printStackTrace();
			} catch (IOException e) {
				e.printStackTrace();
			} 
		}
	}