-
Notifications
You must be signed in to change notification settings - Fork 32
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #151 from Alex-Cook4/rabbitmq
Rabbitmq
- Loading branch information
Showing
8 changed files
with
713 additions
and
459 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
111 changes: 111 additions & 0 deletions
111
...streamsx.messaging/impl/java/src/com/ibm/streamsx/messaging/rabbitmq/AttributeHelper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| /******************************************************************************* | ||
| * Copyright (C) 2015, International Business Machines Corporation | ||
| * All Rights Reserved | ||
| *******************************************************************************/ | ||
| package com.ibm.streamsx.messaging.rabbitmq; | ||
|
|
||
| import java.io.IOException; | ||
| import java.io.InputStream; | ||
| import java.nio.charset.Charset; | ||
| import java.util.Set; | ||
|
|
||
| import com.ibm.streams.operator.Attribute; | ||
| import com.ibm.streams.operator.OutputTuple; | ||
| import com.ibm.streams.operator.StreamSchema; | ||
| import com.ibm.streams.operator.Tuple; | ||
| import com.ibm.streams.operator.Type.MetaType; | ||
| import com.ibm.streams.operator.types.Blob; | ||
| import com.ibm.streams.operator.types.ValueFactory; | ||
|
|
||
| //Helper to check if attributes have been specified explicitly | ||
| class AttributeHelper { | ||
| static final Charset CS = Charset.forName("UTF-8"); | ||
| private boolean wasSet = false, isAvailable = false; | ||
| private MetaType mType = null; | ||
| private String name = null; | ||
| private boolean isString = false; | ||
|
|
||
| AttributeHelper(String n) { | ||
| this.name = n; | ||
| } | ||
|
|
||
| boolean isWasSet() { | ||
| return wasSet; | ||
| } | ||
|
|
||
| boolean isAvailable() { | ||
| return isAvailable; | ||
| } | ||
|
|
||
| String getName() { | ||
| return name; | ||
| } | ||
|
|
||
| void setName(String name) { | ||
| this.name = name; | ||
| wasSet = true; | ||
| } | ||
|
|
||
| void initialize(StreamSchema ss, boolean required, Set<MetaType> supportedTypes) throws Exception { | ||
| Attribute a = ss.getAttribute(name); | ||
| if(a == null) { | ||
| if(wasSet) | ||
| throw new IllegalArgumentException("Attribute \"" + name + "\" not available."); | ||
| if(required) | ||
| throw new IllegalArgumentException("Attribute not found for \"" + name + "\"."); | ||
| return; | ||
| } | ||
| this.mType = a.getType().getMetaType(); | ||
| isString = mType == MetaType.RSTRING || mType == MetaType.USTRING; | ||
|
|
||
| if(!supportedTypes.contains(mType)){ | ||
| throw new Exception("Attribute \"" + name + "\" must be one of: " + supportedTypes); | ||
| } | ||
| isAvailable = true; | ||
| } | ||
| boolean isString() { | ||
| return isString; | ||
| } | ||
|
|
||
| void setValue(OutputTuple otup, String value) { | ||
| if(!isAvailable) return; | ||
| if(isString) { | ||
| if (value == null) | ||
| value = ""; | ||
| otup.setString(name, value); | ||
| } else | ||
| otup.setBlob(name, ValueFactory.newBlob(value.getBytes(CS))); | ||
| } | ||
| void setValue(OutputTuple otup, byte[] value) { | ||
| if(!isAvailable) return; | ||
| if(isString) { | ||
| if (value == null) | ||
| otup.setString(name,""); | ||
| else | ||
| otup.setString(name, new String(value, CS)); | ||
| } | ||
| else | ||
| otup.setBlob(name, ValueFactory.newBlob(value)); | ||
| } | ||
|
|
||
| String getString(Tuple tuple) throws IOException { | ||
| if(!isAvailable) return null; | ||
| if(isString) | ||
| return tuple.getString(name); | ||
| return new String(getBytesFromBlob(tuple, name)); | ||
| } | ||
| byte[] getBytes(Tuple tuple) throws IOException { | ||
| if(!isAvailable) return null; | ||
| if(isString) | ||
| return tuple.getString(name).getBytes(CS); | ||
| return getBytesFromBlob(tuple, name); | ||
| } | ||
| private static byte[] getBytesFromBlob(Tuple tuple, String name) throws IOException { | ||
| Blob blockMsg = tuple.getBlob(name); | ||
| InputStream inputStream = blockMsg.getInputStream(); | ||
| int length = (int) blockMsg.getLength(); | ||
| byte[] byteArray = new byte[length]; | ||
| inputStream.read(byteArray, 0, length); | ||
| return byteArray; | ||
| } | ||
| } |
Oops, something went wrong.