map_support.rb in fluent-plugin-map-0.3.0

- old
+ new

@@ -12,16 +12,21 @@
 #    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 #
 
+require 'fluent/plugin/parser'
+
 module Fluent
   module Plugin
     class MapSupport
       def initialize(map, plugin)
         @map = map
         @plugin = plugin
+        @checker = Fluent::Plugin::Parser::TimeoutChecker.new(@plugin.timeout)
+        @checker.start
+
         if plugin.is_a?(Fluent::Plugin::Filter)
           singleton_class.module_eval(<<-CODE)
           def map_func(time, record)
             #{@map}
           end
@@ -101,15 +106,19 @@
         tuples
       end
 
       def timeout_block
         begin
-          Timeout.timeout(@plugin.timeout){
+          @checker.execute {
             yield
           }
         rescue Timeout::Error
           @plugin.log.error {"Timeout: #{Time.at(time)} #{tag} #{record.inspect}"}
         end
+      end
+
+      def stop
+        @checker.stop
       end
     end
   end
 end