src/twitter/manage/TweetSearchStream.java

   1 package twitter.manage;\r
   2 \r
   3 import java.io.IOException;\r
   4 import java.util.HashMap;\r
   5 import java.util.HashSet;\r
   6 import java.util.Map;\r
   7 import java.util.Set;\r
   8 \r
   9 import twitter.action.streaming.TweetStreamingListener;\r
  10 import twitter4j.DirectMessage;\r
  11 import twitter4j.FilterQuery;\r
  12 import twitter4j.Status;\r
  13 import twitter4j.StatusAdapter;\r
  14 import twitter4j.StatusDeletionNotice;\r
  15 import twitter4j.StatusListener;\r
  16 import twitter4j.StatusStream;\r
  17 import twitter4j.TwitterException;\r
  18 import twitter4j.TwitterStream;\r
  19 import twitter4j.TwitterStreamFactory;\r
  20 import twitter4j.User;\r
  21 import twitter4j.UserStreamAdapter;\r
  22 import twitter4j.auth.AccessToken;\r
  23 \r
  24 /**\r
  25  * 指定したキーワードをサーチする際に利用\r
  26  * @author nishio\r
  27  *\r
  28  */\r
  29 public class TweetSearchStream extends StatusAdapter implements Runnable{\r
  30         //streaming\r
  31         private TwitterStream twitterStream = null;\r
  32         //Filter query\r
  33         private FilterQuery filter = null;\r
  34         //status stream\r
  35         private StatusStream statusStream = null;\r
  36         //Thread\r
  37         private Thread workingThread = null;\r
  38         //tweet manager\r
  39         private TweetManager tweetManager = null;\r
  40         //検索ワードに対応したリスナー\r
  41         private Map<String, TweetStreamingListener> listeners = null;\r
  42         //指定したユーザに対応したリスナー\r
  43         private Map<Long, TweetStreamingListener> userListener = null;\r
  44         //指定したワードの最終更新id\r
  45         private Map<String, Long> lastUpdate = null;\r
  46         //指定したユーザの最終更新id\r
  47         private Map<Long, Long> userLastUpdate = null;\r
  48 \r
  49         /**\r
  50          *\r
  51          * @param consumerKey\r
  52          * @param consumerSecret\r
  53          * @param ac アクセストークン\r
  54          * @param tweetManager\r
  55          */\r
  56         public TweetSearchStream(String consumerKey, String consumerSecret, AccessToken ac, TweetManager tweetManager) {\r
  57                 this.tweetManager = tweetManager;\r
  58                 this.twitterStream = new TwitterStreamFactory().getInstance();\r
  59                 this.twitterStream.setOAuthConsumer(consumerKey, consumerSecret);\r
  60                 this.twitterStream.setOAuthAccessToken(ac);\r
  61 \r
  62                 filter = new FilterQuery();\r
  63                 listeners = new HashMap<String, TweetStreamingListener>();\r
  64                 userListener = new HashMap<Long, TweetStreamingListener>();\r
  65                 lastUpdate = new HashMap<String, Long>();\r
  66                 userLastUpdate = new HashMap<Long, Long>();\r
  67         }\r
  68 \r
  69         /**\r
  70          * 指定した単語を検索対象に加える\r
  71          * @param word\r
  72          * @param listener\r
  73          */\r
  74         public void addSearchWord(String word, TweetStreamingListener listener) {\r
  75                 listeners.put(word, listener);\r
  76                 updateFilter();\r
  77         }\r
  78 \r
  79         /**\r
  80          * 指定した単語を検索対象から外す\r
  81          * @param word\r
  82          */\r
  83         public void removeSearchWord(String word) {\r
  84                 listeners.remove(word);\r
  85                 updateFilter();\r
  86         }\r
  87 \r
  88         /**\r
  89          * 指定したユーザを検索対象に加える\r
  90          * @param userid\r
  91          * @param listener\r
  92          */\r
  93         public void addSearchUser(Long userid, TweetStreamingListener listener) {\r
  94                 userListener.put(userid, listener);\r
  95                 updateFilter();\r
  96         }\r
  97 \r
  98         /**\r
  99          * 指定したユーザを検索対象から外す\r
 100          * @param userid\r
 101          */\r
 102         public void removeSearchUser(Long userid) {\r
 103                 userListener.remove(userid);\r
 104                 updateFilter();\r
 105         }\r
 106 \r
 107         /**\r
 108          * filterの更新\r
 109          */\r
 110         private void updateFilter() {\r
 111                 //指定したユーザの情報を取得するようにする\r
 112                 Long[] users = userListener.keySet().toArray(new Long[0]);\r
 113                 if( users != null ) {\r
 114                         long[] usersLong = new long[users.length];\r
 115                         for(int i=0; i < users.length; i++) {\r
 116                                 usersLong[i] = users[i];\r
 117                         }\r
 118                         filter.follow(usersLong);\r
 119                 }\r
 120                 //指定したワードの情報を取得するようにする\r
 121                 String[] words = listeners.keySet().toArray(new String[0]);\r
 122                 filter.track(words);\r
 123                 workingThread = new Thread(this);\r
 124                 workingThread.start();\r
 125         }\r
 126 \r
 127         /**\r
 128          *指定したワードに対応するステータスを取得\r
 129          */\r
 130         @Override\r
 131         public void onStatus(Status status) {\r
 132                 //検索ワード情報振り分け\r
 133                 Set<String> keys = listeners.keySet();\r
 134                 synchronized (listeners) {\r
 135                         for(String word : keys) {\r
 136                                 if( status.getText().contains( word.toString() ) ) {\r
 137                                         TweetStreamingListener listener = listeners.get(word);\r
 138                                         listener.update(status);\r
 139                                         //最終更新id\r
 140                                         lastUpdate.put(word, status.getId());\r
 141                                 }\r
 142                         }\r
 143                 }\r
 144 \r
 145                 //ユーザ情報振り分け\r
 146                 Set<Long> userKeys = userListener.keySet();\r
 147                 synchronized(userListener) {\r
 148                         for(long id : userKeys) {\r
 149                                 if( status.getUser().getId() == id ) {\r
 150                                         TweetStreamingListener listener = userListener.get(id);\r
 151                                         listener.update(status);\r
 152                                         //最終更新id\r
 153                                         userLastUpdate.put(id, status.getId());\r
 154                                 }\r
 155                         }\r
 156                 }\r
 157         }\r
 158 \r
 159         /**\r
 160          * 最終更新ステータスのidの取得\r
 161          * @param word\r
 162          * @return\r
 163          */\r
 164         public long getLastUpdateID(String word) {\r
 165                 Long id = lastUpdate.get(word);\r
 166                 if( id == null ) {\r
 167                         return 0;\r
 168                 }\r
 169                 return id;\r
 170         }\r
 171 \r
 172         /**\r
 173          * ユーザの最終更新ステータスidの取得\r
 174          * @param userid\r
 175          * @return\r
 176          */\r
 177         public long getUserLastUpdateID(long userid) {\r
 178                 Long id = userLastUpdate.get(userid);\r
 179                 if( id == null ) {\r
 180                         return 0;\r
 181                 }\r
 182                 return id;\r
 183         }\r
 184 \r
 185         /**\r
 186          *\r
 187          */\r
 188         @Override\r
 189         public void onException(Exception ex) {\r
 190                 ex.printStackTrace();\r
 191         }\r
 192 \r
 193         @Override\r
 194         public void run() {\r
 195                 try {\r
 196                         try {\r
 197                                 if( statusStream != null ) {\r
 198                                         //前回開いていたものがあったら閉じる\r
 199                                         statusStream.close();\r
 200                                 }\r
 201                         } catch (IOException e) {\r
 202                                 e.printStackTrace();\r
 203                         }\r
 204                         statusStream = twitterStream.getFilterStream(filter);\r
 205                 } catch (TwitterException e) {\r
 206                         e.printStackTrace();\r
 207                 }\r
 208                 for(; statusStream != null; ) {\r
 209                         try {\r
 210                                 statusStream.next(this);\r
 211                         }catch(Exception e) {\r
 212                                 e.printStackTrace();\r
 213                                 break;\r
 214                         }\r
 215                 }\r
 216         }\r
 217 \r
 218 }\r