12 # after this many changes, a changeset will be closed
15 # this is the scale factor for lat/lon values stored as integers in the database
19 # changeset class keeps some information about changesets downloaded from the
20 # database - enough to let us know which changesets are closed/open & recently
23 attr_reader :id, :created_at, :closed_at, :num_changes
27 @created_at = Time.parse(row["created_at"])
28 @closed_at = Time.parse(row["closed_at"])
29 @num_changes = row["num_changes"].to_i
33 (@closed_at < t) || (@num_changes >= CHANGES_LIMIT)
40 def activity_between?(t1, t2)
41 ((@closed_at >= t1) && (@closed_at < t2)) || ((@created_at >= t1) && (@created_at < t2))
46 # state and connections associated with getting changeset data
47 # replicated to a file.
49 def initialize(config)
50 @config = YAML.load(File.read(config))
51 @state = YAML.load(File.read(@config["state_file"]))
52 @conn = PGconn.connect(@config["db"])
53 @now = Time.now.getutc
57 last_run = @state["last_run"]
58 last_run = (@now - 60) if last_run.nil?
59 @state["last_run"] = @now
60 # pretty much all operations on a changeset will modify its closed_at
61 # time (see rails_port's changeset model). so it is probably enough
62 # for us to look at anything that was closed recently, and filter from
65 .exec("select id, created_at, closed_at, num_changes from changesets where closed_at > ((now() at time zone 'utc') - '1 hour'::interval)")
66 .map { |row| Changeset.new(row) }
67 .select { |cs| cs.activity_between?(last_run, @now) }
69 # set for faster presence lookups by ID
70 cs_ids = Set.new(changesets.map { |c| c.id })
72 # but also add any changesets which have new comments
74 .exec("select distinct changeset_id from changeset_comments where created_at >= '#{last_run}' and created_at < '#{@now}' and visible")
75 .map { |row| row["changeset_id"].to_i }
76 .select { |c_id| not cs_ids.include?(c_id) }
80 .exec("select id, created_at, closed_at, num_changes from changesets where id=#{id}")
81 .map { |row| Changeset.new(row) }
82 .each { |cs| changesets << cs }
85 changesets.sort_by { |cs| cs.id }
88 # creates an XML file containing the changeset information from the
89 # list of changesets output by open_changesets.
90 def changeset_dump(changesets)
91 doc = XML::Document.new
92 doc.root = XML::Node.new("osm")
94 "generator" => "replicate_changesets.rb",
95 "copyright" => "OpenStreetMap and contributors",
96 "attribution" => "http://www.openstreetmap.org/copyright",
97 "license" => "http://opendatacommons.org/licenses/odbl/1-0/" }
98 .each { |k, v| doc.root[k] = v }
100 changesets.each do |cs|
101 xml = XML::Node.new("changeset")
102 xml["id"] = cs.id.to_s
103 xml["created_at"] = cs.created_at.getutc.xmlschema
104 xml["closed_at"] = cs.closed_at.getutc.xmlschema if cs.closed?(@now)
105 xml["open"] = cs.open?(@now).to_s
106 xml["num_changes"] = cs.num_changes.to_s
108 res = @conn.exec("select u.id, u.display_name, c.min_lat, c.max_lat, c.min_lon, c.max_lon from users u join changesets c on u.id=c.user_id where c.id=#{cs.id}")
109 xml["user"] = res[0]["display_name"]
110 xml["uid"] = res[0]["id"]
112 unless res[0]["min_lat"].nil? ||
113 res[0]["max_lat"].nil? ||
114 res[0]["min_lon"].nil? ||
115 res[0]["max_lon"].nil?
116 xml["min_lat"] = (res[0]["min_lat"].to_f / GEO_SCALE).to_s
117 xml["max_lat"] = (res[0]["max_lat"].to_f / GEO_SCALE).to_s
118 xml["min_lon"] = (res[0]["min_lon"].to_f / GEO_SCALE).to_s
119 xml["max_lon"] = (res[0]["max_lon"].to_f / GEO_SCALE).to_s
122 res = @conn.exec("select k, v from changeset_tags where changeset_id=#{cs.id}")
124 tag = XML::Node.new("tag")
130 # grab the visible changeset comments as well
131 res = @conn.exec("select cc.author_id, u.display_name as author, cc.body, cc.created_at from changeset_comments cc join users u on cc.author_id=u.id where cc.changeset_id=#{cs.id} and cc.visible order by cc.created_at asc")
132 xml["comments_count"] = res.num_tuples.to_s
133 if res.num_tuples > 0
134 discussion = XML::Node.new("discussion")
136 comment = XML::Node.new("comment")
137 comment["uid"] = row["author_id"]
138 comment["user"] = row["author"]
139 comment["date"] = Time.parse(row["created_at"]).getutc.xmlschema
140 text = XML::Node.new("text")
141 text.content = row["body"]
143 discussion << comment
154 # saves new state (including the changeset dump xml)
156 File.open(@config["state_file"], "r") do |fl|
157 fl.flock(File::LOCK_EX)
159 sequence = (@state.key?("sequence") ? @state["sequence"] + 1 : 0)
160 data_file = @config["data_dir"] + format("/%03d/%03d/%03d.osm.gz", sequence / 1000000, (sequence / 1000) % 1000, (sequence % 1000))
161 tmp_state = @config["state_file"] + ".tmp"
162 tmp_data = "/tmp/changeset_data.osm.tmp"
163 # try and write the files to tmp locations and then
164 # move them into place later, to avoid in-progress
165 # clashes, or people seeing incomplete files.
167 FileUtils.mkdir_p(File.dirname(data_file))
168 Zlib::GzipWriter.open(tmp_data) do |fh|
169 fh.write(changeset_dump(open_changesets))
171 @state["sequence"] = sequence
172 File.open(tmp_state, "w") do |fh|
173 fh.write(YAML.dump(@state))
175 FileUtils.mv(tmp_data, data_file)
176 FileUtils.mv(tmp_state, @config["state_file"])
177 fl.flock(File::LOCK_UN)
180 STDERR.puts("Error! Couldn't update state.")
181 fl.flock(File::LOCK_UN)
188 rep = Replicator.new(ARGV[0])