Author: igarashitm Date: 2010-10-30 09:58:39 -0400 (Sat, 30 Oct 2010) New Revision: 9824 Modified: trunk/build-hornetq.xml trunk/examples/common/build.xml trunk/examples/core/twitter-connector/build.xml trunk/examples/core/twitter-connector/server0/hornetq-configuration.xml trunk/examples/core/twitter-connector/src/org/hornetq/core/example/TwitterConnectorExample.java trunk/pom.xml trunk/src/main/org/hornetq/integration/twitter/TwitterConstants.java trunk/src/main/org/hornetq/integration/twitter/impl/IncomingTweetsHandler.java trunk/src/main/org/hornetq/integration/twitter/impl/OutgoingTweetsHandler.java trunk/tests/src/org/hornetq/tests/integration/twitter/TwitterTest.java Log: https://jira.jboss.org/browse/HORNETQ-549 * change authentication method from basic auth to OAuth * update twitter4j from 2.1.2 to 2.1.6(now reply-to works collectly) * example's message is now able to be replaced with env.TWITTER_EXAMPLE_ALTERNATIVE_MESSAGE property Modified: trunk/build-hornetq.xml =================================================================== --- trunk/build-hornetq.xml 2010-10-30 07:36:27 UTC (rev 9823) +++ trunk/build-hornetq.xml 2010-10-30 13:58:39 UTC (rev 9824) @@ -1710,8 +1710,10 @@ timeout="${junit.timeout}"> <sysproperty key="user.home" value="${user.home}"/> <sysproperty key="java.io.tmpdir" value="${java.io.tmpdir}"/> - <sysproperty key="twitter.username" value="${twitter.username}"/> - <sysproperty key="twitter.password" value="${twitter.password}"/> + <sysproperty key="twitter.consumerKey" value="${twitter.consumerKey}"/> + <sysproperty key="twitter.consumerSecret" value="${twitter.consumerSecret}"/> + <sysproperty key="twitter.accessToken" value="${twitter.accessToken}"/> + <sysproperty key="twitter.accessTokenSecret" value="${twitter.accessTokenSecret}"/> <jvmarg value="-Djava.library.path=native/bin"/> <jvmarg value="-Dmodule.output=./"/> <jvmarg value="-Djava.util.logging.config.file=src/config/trunk/non-clustered/logging.properties"/> Modified: trunk/examples/common/build.xml =================================================================== --- trunk/examples/common/build.xml 2010-10-30 07:36:27 UTC (rev 9823) +++ trunk/examples/common/build.xml 2010-10-30 13:58:39 UTC (rev 9824) @@ -140,6 +140,7 @@ <!--<echo>client classpath = ${clientClasspath}</echo>--> <property file="${imported.basedir}/config/server.properties"/> <java classname="${example.classname}" fork="true" resultproperty="example-result"> + <jvmarg line="${client.args}"/> <jvmarg value="-Dhornetq.example.server.classpath=${serverclasspath}"/> <jvmarg value="-Dhornetq.example.server.args=${server.args}"/> <jvmarg value="-Dhornetq.example.logserveroutput=${hornetq.example.logserveroutput}"/> Modified: trunk/examples/core/twitter-connector/build.xml =================================================================== --- trunk/examples/core/twitter-connector/build.xml 2010-10-30 07:36:27 UTC (rev 9823) +++ trunk/examples/core/twitter-connector/build.xml 2010-10-30 13:58:39 UTC (rev 9824) @@ -19,10 +19,13 @@ <import file="../../common/build.xml"/> <property environment='env'/> - <target name="check" unless="env.TWITTER_USERNAME"> + <target name="check" unless="env.TWITTER_CONSUMER_KEY"> <echo>**************************************************************************</echo> <echo>* Please set the twitter account: *</echo> - <echo>* ./build.sh -Denv.TWITTER_USERNAME=user -Denv.TWITTER_PASSWORD=password *</echo> + <echo>* ./build.sh -Denv.TWITTER_CONSUMER_KEY=consumerKey \ *</echo> + <echo>* -Denv.TWITTER_CONSUMER_SECRET=consumerSecret \ *</echo> + <echo>* -Denv.TWITTER_ACCESS_TOKEN=accessToken \ *</echo> + <echo>* -Denv.TWITTER_ACCESS_TOKEN_SECRET=accessTokenSecret *</echo> <echo>**************************************************************************</echo> <fail message="run example failed"/> </target> @@ -30,10 +33,11 @@ <target name="run" depends="check"> <antcall target="runExample"> <param name="example.classname" value="org.hornetq.core.example.TwitterConnectorExample"/> + <param name="client.args" value="-Dtwitter.example.alternativeMessage=${env.TWITTER_EXAMPLE_ALTERNATIVE_MESSAGE}"/> <!-- HTTP proxy settings <param name="server.args" value="-Dtwitter4j.http.proxyHost=your.proxy.server -Dtwitter4j.http.proxyPort=your.proxy.port"/> --> - <param name="server.args" value="-Dtwitter.username=${env.TWITTER_USERNAME} -Dtwitter.password=${env.TWITTER_PASSWORD}"/> + <param name="server.args" value="-Dtwitter.consumerKey=${env.TWITTER_CONSUMER_KEY} -Dtwitter.consumerSecret=${env.TWITTER_CONSUMER_SECRET} -Dtwitter.accessToken=${env.TWITTER_ACCESS_TOKEN} -Dtwitter.accessTokenSecret=${env.TWITTER_ACCESS_TOKEN_SECRET}"/> </antcall> </target> Modified: trunk/examples/core/twitter-connector/server0/hornetq-configuration.xml =================================================================== --- trunk/examples/core/twitter-connector/server0/hornetq-configuration.xml 2010-10-30 07:36:27 UTC (rev 9823) +++ trunk/examples/core/twitter-connector/server0/hornetq-configuration.xml 2010-10-30 13:58:39 UTC (rev 9824) @@ -13,7 +13,7 @@ <!-- Acceptors --> <acceptors> <acceptor name="netty-acceptor"> - <factory-class>org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory</factory-class> + <factory-class>org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory</factory-class> </acceptor> </acceptors> @@ -44,15 +44,19 @@ <connector-service name="my-incoming-tweets"> <factory-class>org.hornetq.integration.twitter.TwitterIncomingConnectorServiceFactory</factory-class> <param key="queue" value="queue.incomingQueue"/> - <param key="username" value="${twitter.username}"/> - <param key="password" value="${twitter.password}"/> + <param key="consumerKey" value="${twitter.consumerKey}"/> + <param key="consumerSecret" value="${twitter.consumerSecret}"/> + <param key="accessToken" value="${twitter.accessToken}"/> + <param key="accessTokenSecret" value="${twitter.accessTokenSecret}"/> <param key="interval" value="60"/> </connector-service> <connector-service name="my-outgoing-tweets"> <factory-class>org.hornetq.integration.twitter.TwitterOutgoingConnectorServiceFactory</factory-class> <param key="queue" value="queue.outgoingQueue"/> - <param key="username" value="${twitter.username}"/> - <param key="password" value="${twitter.password}"/> + <param key="consumerKey" value="${twitter.consumerKey}"/> + <param key="consumerSecret" value="${twitter.consumerSecret}"/> + <param key="accessToken" value="${twitter.accessToken}"/> + <param key="accessTokenSecret" value="${twitter.accessTokenSecret}"/> </connector-service> </connector-services> Modified: trunk/examples/core/twitter-connector/src/org/hornetq/core/example/TwitterConnectorExample.java =================================================================== --- trunk/examples/core/twitter-connector/src/org/hornetq/core/example/TwitterConnectorExample.java 2010-10-30 07:36:27 UTC (rev 9823) +++ trunk/examples/core/twitter-connector/src/org/hornetq/core/example/TwitterConnectorExample.java 2010-10-30 13:58:39 UTC (rev 9824) @@ -44,6 +44,11 @@ ClientSession session = null; try { + String testMessage = System.currentTimeMillis() + ": " + System.getProperty("twitter.example.alternativeMessage"); + if(testMessage == null || testMessage.trim().equals("")) { + testMessage = System.currentTimeMillis() + ": ### Hello, HornetQ fans!! We are now experiencing so fast, so reliable and so exciting messaging never seen before ;-) ###"; + } + // Step 1. Create a ClientSessionFactory. csf = HornetQClient.createClientSessionFactory (new TransportConfiguration(NettyConnectorFactory.class.getName())); @@ -58,7 +63,6 @@ // Step 5. Create a core message. ClientMessage cm = session.createMessage(org.hornetq.api.core.Message.TEXT_TYPE,true); - String testMessage = System.currentTimeMillis() + ": ### Hello, HornetQ fans!! We are now experiencing so fast, so reliable and so exciting messaging never seen before ;-) ###"; cm.getBodyBuffer().writeString(testMessage); // Step 6. Send a message to queue.outgoingQueue. @@ -76,13 +80,21 @@ ClientMessage received = cc.receive(70 * 1000); received.acknowledge(); String receivedText = received.getBodyBuffer().readString(); - System.out.println("#### Received a message from " + INCOMING_QUEUE + ": " + receivedText); - if(!receivedText.equals(testMessage)) + while(!receivedText.equals(testMessage)) { - return false; + // ignoring other tweets + received = cc.receiveImmediate(); + if(received == null) { + // no other tweets. test message has gone... + return false; + } + + received.acknowledge(); + receivedText = received.getBodyBuffer().readString(); } - + + System.out.println("#### Received a message from " + INCOMING_QUEUE + ": " + receivedText); return true; } finally Modified: trunk/pom.xml =================================================================== --- trunk/pom.xml 2010-10-30 07:36:27 UTC (rev 9823) +++ trunk/pom.xml 2010-10-30 13:58:39 UTC (rev 9824) @@ -253,7 +253,7 @@ <dependency> <groupId>org.twitter4j</groupId> <artifactId>twitter4j-core</artifactId> - <version>2.1.2</version> + <version>2.1.6</version> </dependency> <!-- needed to compile the tests--> <dependency> Modified: trunk/src/main/org/hornetq/integration/twitter/TwitterConstants.java =================================================================== --- trunk/src/main/org/hornetq/integration/twitter/TwitterConstants.java 2010-10-30 07:36:27 UTC (rev 9823) +++ trunk/src/main/org/hornetq/integration/twitter/TwitterConstants.java 2010-10-30 13:58:39 UTC (rev 9824) @@ -50,32 +50,42 @@ public static final Set<String> ALLOWABLE_OUTGOING_CONNECTOR_KEYS; public static final Set<String> REQUIRED_OUTGOING_CONNECTOR_KEYS; - public static final String USER_NAME = "username"; - public static final String PASSWORD = "password"; + public static final String CONSUMER_KEY = "consumerKey"; + public static final String CONSUMER_SECRET = "consumerSecret"; + public static final String ACCESS_TOKEN ="accessToken"; + public static final String ACCESS_TOKEN_SECRET = "accessTokenSecret"; public static final String QUEUE_NAME = "queue"; public static final String INCOMING_INTERVAL = "interval"; static { ALLOWABLE_INCOMING_CONNECTOR_KEYS = new HashSet<String>(); - ALLOWABLE_INCOMING_CONNECTOR_KEYS.add(USER_NAME); - ALLOWABLE_INCOMING_CONNECTOR_KEYS.add(PASSWORD); + ALLOWABLE_INCOMING_CONNECTOR_KEYS.add(CONSUMER_KEY); + ALLOWABLE_INCOMING_CONNECTOR_KEYS.add(CONSUMER_SECRET); + ALLOWABLE_INCOMING_CONNECTOR_KEYS.add(ACCESS_TOKEN); + ALLOWABLE_INCOMING_CONNECTOR_KEYS.add(ACCESS_TOKEN_SECRET); ALLOWABLE_INCOMING_CONNECTOR_KEYS.add(QUEUE_NAME); ALLOWABLE_INCOMING_CONNECTOR_KEYS.add(INCOMING_INTERVAL); REQUIRED_INCOMING_CONNECTOR_KEYS = new HashSet<String>(); - REQUIRED_INCOMING_CONNECTOR_KEYS.add(USER_NAME); - REQUIRED_INCOMING_CONNECTOR_KEYS.add(PASSWORD); + REQUIRED_INCOMING_CONNECTOR_KEYS.add(CONSUMER_KEY); + REQUIRED_INCOMING_CONNECTOR_KEYS.add(CONSUMER_SECRET); + REQUIRED_INCOMING_CONNECTOR_KEYS.add(ACCESS_TOKEN); + REQUIRED_INCOMING_CONNECTOR_KEYS.add(ACCESS_TOKEN_SECRET); REQUIRED_INCOMING_CONNECTOR_KEYS.add(QUEUE_NAME); ALLOWABLE_OUTGOING_CONNECTOR_KEYS = new HashSet<String>(); - ALLOWABLE_OUTGOING_CONNECTOR_KEYS.add(USER_NAME); - ALLOWABLE_OUTGOING_CONNECTOR_KEYS.add(PASSWORD); + ALLOWABLE_OUTGOING_CONNECTOR_KEYS.add(CONSUMER_KEY); + ALLOWABLE_OUTGOING_CONNECTOR_KEYS.add(CONSUMER_SECRET); + ALLOWABLE_OUTGOING_CONNECTOR_KEYS.add(ACCESS_TOKEN); + ALLOWABLE_OUTGOING_CONNECTOR_KEYS.add(ACCESS_TOKEN_SECRET); ALLOWABLE_OUTGOING_CONNECTOR_KEYS.add(QUEUE_NAME); REQUIRED_OUTGOING_CONNECTOR_KEYS = new HashSet<String>(); - REQUIRED_OUTGOING_CONNECTOR_KEYS.add(USER_NAME); - REQUIRED_OUTGOING_CONNECTOR_KEYS.add(PASSWORD); + REQUIRED_OUTGOING_CONNECTOR_KEYS.add(CONSUMER_KEY); + REQUIRED_OUTGOING_CONNECTOR_KEYS.add(CONSUMER_SECRET); + REQUIRED_OUTGOING_CONNECTOR_KEYS.add(ACCESS_TOKEN); + REQUIRED_OUTGOING_CONNECTOR_KEYS.add(ACCESS_TOKEN_SECRET); REQUIRED_OUTGOING_CONNECTOR_KEYS.add(QUEUE_NAME); } } Modified: trunk/src/main/org/hornetq/integration/twitter/impl/IncomingTweetsHandler.java =================================================================== --- trunk/src/main/org/hornetq/integration/twitter/impl/IncomingTweetsHandler.java 2010-10-30 07:36:27 UTC (rev 9823) +++ trunk/src/main/org/hornetq/integration/twitter/impl/IncomingTweetsHandler.java 2010-10-30 13:58:39 UTC (rev 9824) @@ -23,7 +23,7 @@ import org.hornetq.integration.twitter.TwitterConstants; import org.hornetq.utils.ConfigurationHelper; import twitter4j.*; - +import twitter4j.http.AccessToken; import java.util.Map; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; @@ -39,10 +39,14 @@ private final String connectorName; - private final String userName; + private final String consumerKey; - private final String password; + private final String consumerSecret; + private final String accessToken; + + private final String accessTokenSecret; + private final String queueName; private final int intervalSeconds; @@ -59,17 +63,19 @@ private final ScheduledExecutorService scheduledPool; - private ScheduledFuture scheduledFuture; + private ScheduledFuture<?> scheduledFuture; - public IncomingTweetsHandler(final String connectorName, + public IncomingTweetsHandler(final String connectorName, final Map<String, Object> configuration, final StorageManager storageManager, final PostOffice postOffice, final ScheduledExecutorService scheduledThreadPool) { this.connectorName = connectorName; - this.userName = ConfigurationHelper.getStringProperty(TwitterConstants.USER_NAME, null, configuration); - this.password = ConfigurationHelper.getStringProperty(TwitterConstants.PASSWORD, null, configuration); + this.consumerKey = ConfigurationHelper.getStringProperty(TwitterConstants.CONSUMER_KEY, null, configuration); + this.consumerSecret = ConfigurationHelper.getStringProperty(TwitterConstants.CONSUMER_SECRET, null, configuration); + this.accessToken = ConfigurationHelper.getStringProperty(TwitterConstants.ACCESS_TOKEN, null, configuration); + this.accessTokenSecret = ConfigurationHelper.getStringProperty(TwitterConstants.ACCESS_TOKEN_SECRET, null, configuration); this.queueName = ConfigurationHelper.getStringProperty(TwitterConstants.QUEUE_NAME, null, configuration); Integer intervalSeconds = ConfigurationHelper.getIntProperty(TwitterConstants.INCOMING_INTERVAL, 0, configuration); if (intervalSeconds > 0) @@ -95,9 +101,12 @@ paging = new Paging(); TwitterFactory tf = new TwitterFactory(); - this.twitter = tf.getInstance(userName, password); + this.twitter = tf.getOAuthAuthorizedInstance(this.consumerKey, + this.consumerSecret, + new AccessToken(this.accessToken, + this.accessTokenSecret)); this.twitter.verifyCredentials(); - + // getting latest ID this.paging.setCount(TwitterConstants.FIRST_ATTEMPT_PAGE_SIZE); ResponseList<Status> res = this.twitter.getHomeTimeline(paging); Modified: trunk/src/main/org/hornetq/integration/twitter/impl/OutgoingTweetsHandler.java =================================================================== --- trunk/src/main/org/hornetq/integration/twitter/impl/OutgoingTweetsHandler.java 2010-10-30 07:36:27 UTC (rev 9823) +++ trunk/src/main/org/hornetq/integration/twitter/impl/OutgoingTweetsHandler.java 2010-10-30 13:58:39 UTC (rev 9824) @@ -21,6 +21,7 @@ import org.hornetq.integration.twitter.TwitterConstants; import org.hornetq.utils.ConfigurationHelper; import twitter4j.*; +import twitter4j.http.AccessToken; import java.util.Map; @@ -34,10 +35,14 @@ private final String connectorName; - private final String userName; + private final String consumerKey; - private final String password; + private final String consumerSecret; + private final String accessToken; + + private final String accessTokenSecret; + private final String queueName; private final PostOffice postOffice; @@ -50,13 +55,15 @@ private boolean isStarted = false; - public OutgoingTweetsHandler(final String connectorName, + public OutgoingTweetsHandler(final String connectorName, final Map<String, Object> configuration, final PostOffice postOffice) { this.connectorName = connectorName; - this.userName = ConfigurationHelper.getStringProperty(TwitterConstants.USER_NAME, null, configuration); - this.password = ConfigurationHelper.getStringProperty(TwitterConstants.PASSWORD, null, configuration); + this.consumerKey = ConfigurationHelper.getStringProperty(TwitterConstants.CONSUMER_KEY, null, configuration); + this.consumerSecret = ConfigurationHelper.getStringProperty(TwitterConstants.CONSUMER_SECRET, null, configuration); + this.accessToken = ConfigurationHelper.getStringProperty(TwitterConstants.ACCESS_TOKEN, null, configuration); + this.accessTokenSecret = ConfigurationHelper.getStringProperty(TwitterConstants.ACCESS_TOKEN_SECRET, null, configuration); this.queueName = ConfigurationHelper.getStringProperty(TwitterConstants.QUEUE_NAME, null, configuration); this.postOffice = postOffice; } @@ -91,8 +98,12 @@ this.queue = (Queue)b.getBindable(); TwitterFactory tf = new TwitterFactory(); - this.twitter = tf.getInstance(userName, password); + this.twitter = tf.getOAuthAuthorizedInstance(this.consumerKey, + this.consumerSecret, + new AccessToken(this.accessToken, + this.accessTokenSecret)); this.twitter.verifyCredentials(); + // TODO make filter-string configurable // this.filter = FilterImpl.createFilter(filterString); this.filter = null; Modified: trunk/tests/src/org/hornetq/tests/integration/twitter/TwitterTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/twitter/TwitterTest.java 2010-10-30 07:36:27 UTC (rev 9823) +++ trunk/tests/src/org/hornetq/tests/integration/twitter/TwitterTest.java 2010-10-30 13:58:39 UTC (rev 9824) @@ -38,6 +38,7 @@ import org.hornetq.tests.util.ServiceTestBase; import org.hornetq.tests.util.UnitTestCase; import twitter4j.*; +import twitter4j.http.AccessToken; /** * A TwitterTest @@ -50,19 +51,23 @@ { private static final Logger log = Logger.getLogger(TwitterTest.class); private static final String KEY_CONNECTOR_NAME = "connector.name"; - private static final String KEY_USERNAME = "username"; - private static final String KEY_PASSWORD = "password"; + private static final String KEY_CONSUMER_KEY = "consumerKey"; + private static final String KEY_CONSUMER_SECRET = "consumerSecret"; + private static final String KEY_ACCESS_TOKEN = "accessToken"; + private static final String KEY_ACCESS_TOKEN_SECRET = "accessTokenSecret"; private static final String KEY_QUEUE_NAME = "queue.name"; - private static final String TWITTER_USERNAME = System.getProperty("twitter.username"); - private static final String TWITTER_PASSWORD = System.getProperty("twitter.password"); - + private static final String TWITTER_CONSUMER_KEY = System.getProperty("twitter.consumerKey"); + private static final String TWITTER_CONSUMER_SECRET = System.getProperty("twitter.consumerSecret"); + private static final String TWITTER_ACCESS_TOKEN = System.getProperty("twitter.accessToken"); + private static final String TWITTER_ACCESS_TOKEN_SECRET = System.getProperty("twitter.accessTokenSecret"); + @Override protected void setUp() throws Exception { - if(TWITTER_USERNAME == null || TWITTER_PASSWORD == null) + if(TWITTER_CONSUMER_KEY == null || TWITTER_CONSUMER_SECRET == null || TWITTER_ACCESS_TOKEN == null || TWITTER_ACCESS_TOKEN_SECRET == null) { - throw new Exception("* * * Please set twitter.username and twitter.password in system property * * *"); + throw new Exception("* * * Please set twitter.consumerKey, twitter.consumerSecret, twitter.accessToken and twitter.accessTokenSecuret in system property * * *"); } super.setUp(); } @@ -101,8 +106,10 @@ public void testIncomingWithInvalidCredentials() throws Exception { HashMap<String,String> params = new HashMap<String,String>(); - params.put(KEY_USERNAME, "invalidUsername"); - params.put(KEY_PASSWORD, "invalidPassword"); + params.put(KEY_CONSUMER_KEY, "invalidConsumerKey"); + params.put(KEY_CONSUMER_SECRET, "invalidConsumerSecret"); + params.put(KEY_ACCESS_TOKEN, "invalidAccessToken"); + params.put(KEY_ACCESS_TOKEN_SECRET, "invalidAcccessTokenSecret"); internalTestIncomingFailedToInitialize(params); } @@ -139,18 +146,14 @@ public void testOutgoingWithInvalidCredentials() throws Exception { HashMap<String,String> params = new HashMap<String,String>(); - params.put(KEY_USERNAME, "invalidUsername"); - params.put(KEY_PASSWORD, "invalidPassword"); + params.put(KEY_CONSUMER_KEY, "invalidConsumerKey"); + params.put(KEY_CONSUMER_SECRET, "invalidConsumerSecret"); + params.put(KEY_ACCESS_TOKEN, "invalidAccessToken"); + params.put(KEY_ACCESS_TOKEN_SECRET, "invalidAcccessTokenSecret"); internalTestOutgoingFailedToInitialize(params); } - /** - * This will fail until TFJ-347 is fixed. - * http://twitter4j.org/jira/browse/TFJ-347 - * - * @throws Exception - */ - public void _testOutgoingWithInReplyTo() throws Exception + public void testOutgoingWithInReplyTo() throws Exception { internalTestOutgoingWithInReplyTo(); } @@ -161,7 +164,10 @@ ClientSession session = null; String queue = "TwitterTestQueue"; int interval = 5; - Twitter twitter = new TwitterFactory().getInstance(TWITTER_USERNAME,TWITTER_PASSWORD); + Twitter twitter = new TwitterFactory().getOAuthAuthorizedInstance(TWITTER_CONSUMER_KEY, + TWITTER_CONSUMER_SECRET, + new AccessToken(TWITTER_ACCESS_TOKEN, + TWITTER_ACCESS_TOKEN_SECRET)); String testMessage = "TwitterTest/incoming: " + System.currentTimeMillis(); log.debug("test incoming: " + testMessage); @@ -171,8 +177,10 @@ HashMap<String, Object> config = new HashMap<String, Object>(); config.put(TwitterConstants.INCOMING_INTERVAL, interval); config.put(TwitterConstants.QUEUE_NAME, queue); - config.put(TwitterConstants.USER_NAME, TWITTER_USERNAME); - config.put(TwitterConstants.PASSWORD, TWITTER_PASSWORD); + config.put(TwitterConstants.CONSUMER_KEY, TWITTER_CONSUMER_KEY); + config.put(TwitterConstants.CONSUMER_SECRET, TWITTER_CONSUMER_SECRET); + config.put(TwitterConstants.ACCESS_TOKEN, TWITTER_ACCESS_TOKEN); + config.put(TwitterConstants.ACCESS_TOKEN_SECRET, TWITTER_ACCESS_TOKEN_SECRET); ConnectorServiceConfiguration inconf = new ConnectorServiceConfiguration( TwitterIncomingConnectorServiceFactory.class.getName(), @@ -244,22 +252,32 @@ HornetQServer server0 = null; String connectorName = "test-incoming-connector"; String queue = "TwitterTestQueue"; - String userName = "invalidUsername"; - String password = "invalidPassword"; + String consumerKey = "invalidConsumerKey"; + String consumerSecret = "invalidConsumerSecret"; + String accessToken = "invalidAccessToken"; + String accessTokenSecret = "invalidAccessTokenSecret"; int interval = 5; if(params.containsKey(KEY_CONNECTOR_NAME)) { connectorName = params.get(KEY_CONNECTOR_NAME); } - if(params.containsKey(KEY_USERNAME)) + if(params.containsKey(KEY_CONSUMER_KEY)) { - userName = params.get(KEY_USERNAME); + consumerKey = params.get(KEY_CONSUMER_KEY); } - if(params.containsKey(KEY_PASSWORD)) + if(params.containsKey(KEY_CONSUMER_SECRET)) { - password = params.get(KEY_PASSWORD); + consumerSecret = params.get(KEY_CONSUMER_SECRET); } + if(params.containsKey(KEY_ACCESS_TOKEN)) + { + accessToken = params.get(KEY_ACCESS_TOKEN); + } + if(params.containsKey(KEY_ACCESS_TOKEN_SECRET)) + { + accessTokenSecret = params.get(KEY_ACCESS_TOKEN_SECRET); + } if(params.containsKey(KEY_QUEUE_NAME)) { queue = params.get(KEY_QUEUE_NAME); @@ -271,8 +289,10 @@ HashMap<String, Object> config = new HashMap<String, Object>(); config.put(TwitterConstants.INCOMING_INTERVAL, interval); config.put(TwitterConstants.QUEUE_NAME, queue); - config.put(TwitterConstants.USER_NAME, userName); - config.put(TwitterConstants.PASSWORD, password); + config.put(TwitterConstants.CONSUMER_KEY, consumerKey); + config.put(TwitterConstants.CONSUMER_SECRET, consumerSecret); + config.put(TwitterConstants.ACCESS_TOKEN, accessToken); + config.put(TwitterConstants.ACCESS_TOKEN_SECRET, accessTokenSecret); ConnectorServiceConfiguration inconf = new ConnectorServiceConfiguration(TwitterIncomingConnectorServiceFactory.class.getName(), config, @@ -306,7 +326,10 @@ HornetQServer server0 = null; ClientSession session = null; String queue = "TwitterTestQueue"; - Twitter twitter = new TwitterFactory().getInstance(TWITTER_USERNAME,TWITTER_PASSWORD); + Twitter twitter = new TwitterFactory().getOAuthAuthorizedInstance(TWITTER_CONSUMER_KEY, + TWITTER_CONSUMER_SECRET, + new AccessToken(TWITTER_ACCESS_TOKEN, + TWITTER_ACCESS_TOKEN_SECRET)); String testMessage = "TwitterTest/outgoing: " + System.currentTimeMillis(); log.debug("test outgoing: " + testMessage); @@ -315,8 +338,10 @@ Configuration configuration = createDefaultConfig(false); HashMap<String, Object> config = new HashMap<String, Object>(); config.put(TwitterConstants.QUEUE_NAME, queue); - config.put(TwitterConstants.USER_NAME, TWITTER_USERNAME); - config.put(TwitterConstants.PASSWORD, TWITTER_PASSWORD); + config.put(TwitterConstants.CONSUMER_KEY, TWITTER_CONSUMER_KEY); + config.put(TwitterConstants.CONSUMER_SECRET, TWITTER_CONSUMER_SECRET); + config.put(TwitterConstants.ACCESS_TOKEN, TWITTER_ACCESS_TOKEN); + config.put(TwitterConstants.ACCESS_TOKEN_SECRET, TWITTER_ACCESS_TOKEN_SECRET); ConnectorServiceConfiguration outconf = new ConnectorServiceConfiguration(TwitterOutgoingConnectorServiceFactory.class.getName(), config, @@ -388,25 +413,35 @@ protected void internalTestOutgoingFailedToInitialize(HashMap<String,String> params) throws Exception { HornetQServer server0 = null; - String connectorName = "test-outgoing-connector"; + String connectorName = "test-outgoing-connector"; String queue = "TwitterTestQueue"; - String userName = TWITTER_USERNAME; - String password = TWITTER_PASSWORD; + String consumerKey = TWITTER_CONSUMER_KEY; + String consumerSecret = TWITTER_CONSUMER_SECRET; + String accessToken = TWITTER_ACCESS_TOKEN; + String accessTokenSecret = TWITTER_ACCESS_TOKEN_SECRET; if(params.containsKey(KEY_CONNECTOR_NAME)) { connectorName = params.get(KEY_CONNECTOR_NAME); } - if(params.containsKey(KEY_USERNAME)) + if (params.containsKey(KEY_CONSUMER_KEY)) { - userName = params.get(KEY_USERNAME); + consumerKey = params.get(KEY_CONSUMER_KEY); } - if(params.containsKey(KEY_PASSWORD)) + if (params.containsKey(KEY_CONSUMER_SECRET)) { - password = params.get(KEY_PASSWORD); + consumerSecret = params.get(KEY_CONSUMER_SECRET); } - if(params.containsKey(KEY_QUEUE_NAME)) + if (params.containsKey(KEY_ACCESS_TOKEN)) { + accessToken = params.get(KEY_ACCESS_TOKEN); + } + if (params.containsKey(KEY_ACCESS_TOKEN_SECRET)) + { + accessTokenSecret = params.get(KEY_ACCESS_TOKEN_SECRET); + } + if (params.containsKey(KEY_QUEUE_NAME)) + { queue = params.get(KEY_QUEUE_NAME); } @@ -415,12 +450,14 @@ Configuration configuration = createDefaultConfig(false); HashMap<String, Object> config = new HashMap<String, Object>(); config.put(TwitterConstants.QUEUE_NAME, queue); - config.put(TwitterConstants.USER_NAME, userName); - config.put(TwitterConstants.PASSWORD, password); + config.put(TwitterConstants.CONSUMER_KEY, consumerKey); + config.put(TwitterConstants.CONSUMER_SECRET, consumerSecret); + config.put(TwitterConstants.ACCESS_TOKEN, accessToken); + config.put(TwitterConstants.ACCESS_TOKEN_SECRET, accessTokenSecret); ConnectorServiceConfiguration outconf = new ConnectorServiceConfiguration(TwitterOutgoingConnectorServiceFactory.class.getName(), config, - "test-outgoing-connector"); + connectorName); configuration.getConnectorServiceConfigurations().add(outconf); CoreQueueConfiguration qc = new CoreQueueConfiguration(queue, queue, null, false); configuration.getQueueConfigurations().add(qc); @@ -446,16 +483,21 @@ HornetQServer server0 = null; ClientSession session = null; String queue = "TwitterTestQueue"; - Twitter twitter = new TwitterFactory().getInstance(TWITTER_USERNAME,TWITTER_PASSWORD); + Twitter twitter = new TwitterFactory().getOAuthAuthorizedInstance(TWITTER_CONSUMER_KEY, + TWITTER_CONSUMER_SECRET, + new AccessToken(TWITTER_ACCESS_TOKEN, + TWITTER_ACCESS_TOKEN_SECRET)); String testMessage = "TwitterTest/outgoing with in_reply_to: " + System.currentTimeMillis(); - String replyMessage = "@" + TWITTER_USERNAME + " TwitterTest/outgoing reply: " + System.currentTimeMillis(); + String replyMessage = "@" + twitter.getScreenName() + " TwitterTest/outgoing reply: " + System.currentTimeMillis(); try { Configuration configuration = createDefaultConfig(false); HashMap<String, Object> config = new HashMap<String, Object>(); config.put(TwitterConstants.QUEUE_NAME, queue); - config.put(TwitterConstants.USER_NAME, TWITTER_USERNAME); - config.put(TwitterConstants.PASSWORD, TWITTER_PASSWORD); + config.put(TwitterConstants.CONSUMER_KEY, TWITTER_CONSUMER_KEY); + config.put(TwitterConstants.CONSUMER_SECRET, TWITTER_CONSUMER_SECRET); + config.put(TwitterConstants.ACCESS_TOKEN, TWITTER_ACCESS_TOKEN); + config.put(TwitterConstants.ACCESS_TOKEN_SECRET, TWITTER_ACCESS_TOKEN_SECRET); ConnectorServiceConfiguration outconf = new ConnectorServiceConfiguration(TwitterOutgoingConnectorServiceFactory.class.getName(), config,