1 /*
2 * Licensed to the Apache Software Foundation (ASF) under one
3 * or more contributor license agreements. See the NOTICE file
4 * distributed with this work for additional information
5 * regarding copyright ownership. The ASF licenses this file
6 * to you under the Apache License, Version 2.0 (the
7 * "License"); you may not use this file except in compliance
8 * with the License. You may obtain a copy of the License at
9 *
10 * http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing,
13 * software distributed under the License is distributed on an
14 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 * KIND, either express or implied. See the License for the
16 * specific language governing permissions and limitations
17 * under the License.
18 *
19 */
20 package org.apache.directory.server.ldap.replication.consumer;
21
22
23 import java.util.Queue;
24 import java.util.concurrent.ConcurrentLinkedQueue;
25
26 import org.apache.directory.api.ldap.model.constants.Loggers;
27 import org.slf4j.Logger;
28 import org.slf4j.LoggerFactory;
29 import org.slf4j.MDC;
30
31
32 /**
33 * A thread used to ping the provider o check if they are alive or not.
34 *
35 * @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
36 */
37 public class PingerThread extends Thread
38 {
39 /** Logger for the replication consumer */
40 private static final Logger CONSUMER_LOG = LoggerFactory.getLogger( Loggers.CONSUMER_LOG.getName() );
41
42 /** The list of consumers we want to check */
43 private Queue<ReplicationConsumer> consumers = new ConcurrentLinkedQueue<>();
44
45 /** A flag to stop the pinger */
46 private boolean stop = false;
47
48 /** the time interval before this thread pings each replication provider. Default value is 5 seconds */
49 private long sleepTime = 5000;
50
51 /**
52 * Create a new instance of this thread.
53 *
54 * @param sleepSec the number of seconds pinger thread should sleep before pinging the providers
55 */
56 public PingerThread( int sleepSec )
57 {
58 if ( sleepSec > 0 )
59 {
60 sleepTime = sleepSec * 1000L;
61 }
62
63 CONSUMER_LOG.info( "Configured pinger thread to sleep for {} seconds", ( sleepTime / 1000 ) );
64
65 setDaemon( true );
66 }
67
68
69 /**
70 * Starts the thread
71 */
72 @Override
73 public void run()
74 {
75 try
76 {
77 if ( CONSUMER_LOG.isDebugEnabled() )
78 {
79 MDC.put( "Replica", "Pinger" );
80
81 CONSUMER_LOG.debug( "Starting the provider's pinger" );
82 }
83
84 while ( !stop )
85 {
86 for ( ReplicationConsumer consumer : consumers )
87 {
88 consumer.ping();
89 }
90
91 Thread.sleep( sleepTime );
92 }
93 }
94 catch ( InterruptedException ie )
95 {
96 CONSUMER_LOG.debug( "The pinger has been interrupted" );
97 }
98 }
99
100
101 /**
102 * Add a new consumer to ping
103 *
104 * @param consumer The consumer we want to ping
105 */
106 public void addConsumer( ReplicationConsumer consumer )
107 {
108 if ( !consumers.contains( consumer ) )
109 {
110 consumers.add( consumer );
111 }
112 }
113
114
115 /**
116 * Remove a consumer to ping
117 * @param consumer The consumer we want to remove
118 */
119 public void removeConsumer( ReplicationConsumer consumer )
120 {
121 consumers.remove( consumer );
122 }
123
124
125 /**
126 * Stops the ping for all the consumers
127 */
128 public void stopPinging()
129 {
130 stop = true;
131 }
132 }