AbstractService


1   package org.jgroups.service;
2   
3   import org.apache.commons.logging.Log;
4   import org.apache.commons.logging.LogFactory;
5   import org.jgroups.*;
6   import org.jgroups.blocks.PullPushAdapter;
7   
8   import java.util.LinkedList  ;
9   
10  /**
11   * <code>AbstractService</code> is a superclass for all service implementations.
12   * Each service has two communication channels: one for inter-service 
13   * communication with group memebership enabled, and one for client communication
14   * witout group membership. This allows creation of full featured services with
15   * replication and load balancing and light-weight clients that can call service 
16   * methods by sending specific messages.
17   *
18   * @author Roman Rokytskyy (rrokytskyy@acm.org)
19   */
20  public abstract class AbstractService implements MembershipListener {
21  
22      protected final Channel serviceChannel;
23  
24      protected final PullPushAdapter serviceAdapter;
25      
26      protected final Channel clientChannel;
27      
28      protected final LinkedList   members = new LinkedList  ();
29  
30      protected boolean blocked;
31  
32      // this monitor is used to simplify subclasses to stop
33      // inter-service communication when new view is being
34      // installed. Simple blockMonitor.wait() will stop sending
35      // thread until new view is accepted.
36      protected final Object   blockMonitor = new Object  ();
37      
38      // this monitor is used to start service as standalone
39      // application if value of runThread is true
40      protected final Object   threadMonitor = new Object  ();
41      protected boolean runThread;
42  
43      protected final Log log=LogFactory.getLog(this.getClass());
44  
45  
46      /**
47       * Main constructor to create services. It creates instance of service
48       * class, registers {@link PullPushAdapter} for inter-service channel and
49       * tries to retrive initial service state.
50       * 
51       * @param serviceChannel instance of {@link Channel} class that will be
52       * used for inter-service communication. This channel must provide group
53       * membership service and reliable group multicast.
54       * 
55       * @param clientChannel instance of {@link Channel} class that will be used
56       * for client-service communication. This channel should provide reliable
57       * group multicast and unicast, but specific service implementation might
58       * put weaker requirements.
59       */
60      public AbstractService(Channel serviceChannel, Channel clientChannel) {
61          this.serviceChannel = serviceChannel;
62          this.serviceAdapter = new PullPushAdapter(serviceChannel, this);
63          this.clientChannel = clientChannel;
64          
65          try {
66              serviceChannel.getState(null, 1000);
67          }
68          catch(Exception   ex) {
69              if(log.isErrorEnabled()) log.error("exception fetching state: " + ex);
70          }
71          
72      }
73      
74      /**
75       * Get name of this service. Name of the service determines service type,
76       * but not an instance. Instance of the service is uniquely determined
77       * by {@link #getAddress()} value.
78       * 
79       * @return name of this service.
80       */
81      public abstract String   getName();
82      
83      /**
84       * Set message listener for service message channel.
85       */
86      protected void setMessageListener(MessageListener listener) {
87      this.serviceAdapter.setListener(listener);
88      }
89  
90      /**
91       * Get address of this service in service group.
92       */
93      public Address getAddress() {
94      return serviceChannel.getLocalAddress();
95      }
96  
97      /**
98       * Check if this service is a coordinator of service group.
99       *
100      * @return <code>true</code> if this service is a coordinator of service
101      * group.
102      */
103     public boolean isCoordinator() {
104     return getAddress().equals(members.getFirst());
105     }
106 
107     /**
108      * This method returns <code>true</code> if inter-service communication
109      * processes should temporarily stop sending messages to service channel.
110      */
111     public boolean isBlocked() {
112     return blocked;
113     }
114 
115     /**
116      * Stop current thread's execution until inter-service channel is unblocked.
117      * This is a service method to simplify development of channel management
118      * in subclasses.
119      */
120     public void waitOnBlocked() throws InterruptedException   {
121     synchronized(blockMonitor) {
122         blockMonitor.wait();
123     }
124     }
125 
126     /**
127      * This method is called when service is supposed to stop sending messages
128      * to channel until new view is installed. Services that use inter-service
129      * communication should respect this method.
130      */
131     public void block() {
132     blocked = true;
133     }
134 
135     /**
136      * This method is called when a member of service group is suspected to
137      * be failed.
138      */
139     public void suspect(Address suspectedMember) {
140     }
141 
142     /**
143      * This method is called when new view is installed. We make local copy
144      * of view members. If any other information is needed, it can be accessed
145      * via {@link #getChannel()} method.
146      *
147      * @param view new view that was accepted.
148      */
149     public void viewAccepted(View view) {
150     synchronized(members) {
151         members.clear();
152         members.addAll(view.getMembers());
153     }
154 
155     // notify all waiting threads
156     synchronized(blockMonitor) {
157         blocked = false;
158         blockMonitor.notifyAll();
159     }
160     }
161     
162     /**
163      * Start standalone thread that will run until explicitly stopped. This 
164      * allows running this service as standalone process.
165      */
166     public void start() {
167     runThread = true;
168     
169     Runnable   runnable = new Runnable  () {
170             public void run() {
171         while(runThread) 
172             synchronized(threadMonitor) {
173             try {
174                 threadMonitor.wait();
175             } catch(InterruptedException   ex) {
176             }
177             }
178             }
179     };
180     
181     Thread   thread = new Thread  (runnable, getName() + " Thread [" + getAddress() + ']');
182         // thread.setDaemon(true);
183     thread.start();
184     }
185     
186     /**
187      * Stop standalone thread started with {@link start()} method. If no thread
188      * were started this method does nothing.
189      */
190     public void stop() {
191     if (!runThread)
192         return;
193         
194     runThread = false;
195     
196     synchronized(threadMonitor) {
197         threadMonitor.notifyAll();
198     }
199     }
200 
201 }
202
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags